Update srx-2025 start scaffolding for pydantic-eda 0.4.0

.gitignore

@@ -169,3 +169,5 @@ cython_debug/
 
 # PyPI configuration file
 .pypirc
+
+.vscode

main.py

@@ -6,7 +6,11 @@
 
 
 def main():
-    eda = EDAClient(base_url="https://<your_group_id>.srexperts.net:9443")
+    eda = EDAClient(
+        base_url="https://<your_group_id>.srexperts.net:9443",
+        username="admin",
+        password="<fill in the password>",
+    )
 
     # the virtualnetwork function returns an empty object. Fill it in.
     my_virtualnetwork = virtualnetwork(ns="eda", name="my-vnet-using-python")

pyproject.toml

@@ -12,4 +12,4 @@ dependencies = [
 ]
 
 [tool.uv.sources]
-pydantic-eda = { git = "https://github.com/eda-labs/pydantic-eda", tag = "v0.3.2" }
+pydantic-eda = { git = "https://github.com/eda-labs/pydantic-eda", tag = "v0.4.0-rc1" }

src/client.py

@@ -1,16 +1,17 @@
 import logging
-from typing import Any, List, Literal, Optional
+from typing import Any, List, Literal, Optional, Tuple
 
 import httpx
 from pydantic import BaseModel
-from pydantic_eda.core.v25_4_1.models import (
+from pydantic_eda.core.v25_8_1_rc1.models import (
     GroupVersionKind,
     NsCrGvkName,
     Transaction,
     TransactionContent,
     TransactionCr,
-    TransactionDetails,
+    TransactionExecutionResult,
     TransactionId,
+    TransactionSummaryResult,
     TransactionType,
     TransactionValue,
 )
@@ -24,25 +25,28 @@
 
 logger = logging.getLogger(__name__)
 
-EDA_VERSION = "v25.4.1"
-
 KC_REALM = "master"
 KC_CLIENT_ID = "admin-cli"
 KC_USERNAME = "admin"
 KC_PASSWORD = "admin"
 EDA_REALM = "eda"
 API_CLIENT_ID = "eda"
 
+KC_API_URL = "/core/httpproxy/v1/keycloak"
+TX_API_URL = "/core/transaction/v2"
+
 
 class EDAClient(httpx.Client):
-    def __init__(self, base_url: str):
+    def __init__(self, base_url: str, username: str, password: str):
         self.base_url: str = base_url
-        self.kc_url: str = self.base_url.join("/core/httpproxy/v1/keycloak")
+        self.kc_url: str = self.base_url.join(KC_API_URL)
+        self.username: str = username
+        self.password: str = password
 
         self.headers: dict[str, str] = {}
         self.token: str = ""
         self.transaction: Optional[Transaction] = None
-        self.transaction_endpoint: str = self.base_url.join("/core/transaction/v1")
+        self.transaction_endpoint: str = self.base_url.join(TX_API_URL)
 
         super().__init__(headers=self.headers, verify=False)
 
@@ -131,7 +135,7 @@ def add_to_transaction(self, resource: BaseModel, type: TxType) -> None:
             self.transaction.crs.append(tx_cr)
 
     def commit_transaction(self) -> Any:
-        """Commit transaction"""
+        """Commit transaction and get its status"""
 
         # convert the transaction instance to a dict
         if self.transaction is None:
@@ -140,15 +144,15 @@ def commit_transaction(self) -> Any:
         self.transaction.retain = True
         self.transaction.resultType = "normal"
 
-        content = self.transaction.model_dump_json(
+        tx_content = self.transaction.model_dump_json(
             exclude_unset=True, exclude_none=True, exclude_defaults=True
         )
 
         # logger.info(f"Committing transaction: {content}")
 
         response = self.post(
             url=self.transaction_endpoint,
-            content=content,
+            content=tx_content,
         )
         if response.status_code != 200:
             raise ValueError(response.text)
@@ -159,29 +163,23 @@ def commit_transaction(self) -> Any:
 
         logger.info(f"Transaction {tx_id.id} committed")
 
-        tx_details: TransactionDetails = self.get_transaction_details(tx_id.id)
-        errs = self.tx_must_succeed(tx_details)
+        tx_summary_result, errs = self.tx_must_succeed(tx_id.id)
         if errs:
             logger.error(f"Transaction {tx_id.id} errors:\n  - " + "\n  - ".join(errs))
-        logger.info(f"Transaction {tx_id.id} state: {tx_details.state}")
+        logger.info(f"Transaction {tx_id.id} state: {tx_summary_result.state}")
 
-    def get_transaction_details(self, tx_id: int) -> TransactionDetails:
-        """Get transaction"""
+    def get_transaction_summary_result(self, tx_id: int) -> TransactionSummaryResult:
+        """Get transaction summary result"""
 
-        params = {
-            "waitForComplete": "true",
-            # "failOnErrors": "true"
-        }
         response = self.get(
-            url=f"{self.transaction_endpoint}/details/{tx_id}",
-            params=params,
+            url=f"{self.transaction_endpoint}/result/summary/{tx_id}",
         )
         if response.status_code != 200:
             raise ValueError(response.text)
 
         # logger.info(f"Transaction {tx_id} details:\n{response.json()}")
 
-        return TransactionDetails(**response.json())
+        return TransactionSummaryResult(**response.json())
 
     def _get_client_secret(self) -> str:
         client = self
@@ -238,8 +236,8 @@ def _get_access_token(self) -> str:
             "client_id": API_CLIENT_ID,
             "grant_type": "password",
             "scope": "openid",
-            "username": "admin",
-            "password": "admin",
+            "username": self.username,
+            "password": self.password,
             "client_secret": client_secret,
         }
 
@@ -249,7 +247,45 @@ def _get_access_token(self) -> str:
         response.raise_for_status()
         return response.json()["access_token"]
 
-    def tx_must_succeed(self, tx_details: TransactionDetails) -> List[str] | None:
-        """Check if transaction succeeded"""
-        if not tx_details.success:
-            return tx_details.generalErrors
+    def tx_must_succeed(
+        self, tx_id: int
+    ) -> Tuple[TransactionSummaryResult, Optional[List[str]]]:
+        """Wait for transaction to complete and fetch its status and execution errors"""
+
+        errors: List[str] = []
+
+        tx_exec_result: TransactionExecutionResult = self.get_transaction_exec_result(
+            tx_id
+        )
+
+        tx_summary_result: TransactionSummaryResult = (
+            self.get_transaction_summary_result(tx_id)
+        )
+
+        if not tx_summary_result.success:
+            if tx_exec_result.generalErrors:
+                errors.extend(tx_exec_result.generalErrors)
+            # find errors in the intents
+            if tx_exec_result.intentsRun:
+                for intent_result in tx_exec_result.intentsRun:
+                    if intent_result.errors:
+                        for error in intent_result.errors:
+                            errors.append(
+                                f"Intent '{intent_result.intentName}' error: {error.rawError}"
+                            )
+
+        return (tx_summary_result, errors)
+
+    def get_transaction_exec_result(self, tx_id: int) -> TransactionExecutionResult:
+        """Get transaction execution result and wait for transaction to complete"""
+        params = {"waitForComplete": "true"}
+        response = self.get(
+            url=f"{self.transaction_endpoint}/result/execution/{tx_id}", params=params
+        )
+
+        if response.status_code != 200:
+            raise ValueError(response.text)
+
+        # logger.info(f"Transaction {tx_id} exec results:\n{response.json()}")
+
+        return TransactionExecutionResult(**response.json())

src/virtualnetwork.py

@@ -1,4 +1,4 @@
-import pydantic_eda.apps.services.v1alpha1.models as service
+import pydantic_eda.apps.services.v1.models as service
 
 
 def virtualnetwork(ns: str, name: str) -> service.VirtualNetwork: