async bs

Author: hellt

.python-version

@@ -1 +1 @@
-3.13
+3.12

hello.py

@@ -1,26 +1,44 @@
-from models.com.nokia.eda.interfaces.v1alpha1 import Interface, Member, SpecModel, Metadata
+import asyncio
+
+from models.com.nokia.eda.interfaces.v1alpha1 import (
+    Interface,
+    Member,
+    Metadata,
+    SpecModel,
+)
+from src import auth
+
+
+async def get_api_token():
+    client_secret = await auth.get_client_secret()
+    access_token = await auth.get_access_token(client_secret)
+    print(f"Access Token: {access_token}")
+
 
 def main():
-    iface = Interface(
-        apiVersion="interfaces.eda.nokia.com/v1alpha1",
-        kind="Interface",
-        metadata=Metadata(
-            name="test",
-            namespace="clab-vlan",
-            labels={"app": "test"},
-        ),
-        spec=SpecModel(
-            description="This is a test interface",
-            enabled=True,
-            members=[
-                Member(
-                    node="leaf1",
-                    interface="ethernet-1-1",
-                )
-            ]
-        )
-    )
-    print(iface.model_dump_json(indent=2))
+    # Run the asynchronous function
+    asyncio.run(get_api_token())
+
+    # iface = Interface(
+    #     apiVersion="interfaces.eda.nokia.com/v1alpha1",
+    #     kind="Interface",
+    #     metadata=Metadata(
+    #         name="test",
+    #         namespace="clab-vlan",
+    #         labels={"app": "test"},
+    #     ),
+    #     spec=SpecModel(
+    #         description="This is a test interface",
+    #         enabled=True,
+    #         members=[
+    #             Member(
+    #                 node="leaf1",
+    #                 interface="ethernet-1-1",
+    #             )
+    #         ]
+    #     )
+    # )
+    # print(iface.model_dump_json(indent=2))
 
 
 if __name__ == "__main__":

pyproject.toml

@@ -1,9 +1,7 @@
 [project]
 name = "openapi-example-python"
 version = "0.1.0"
-description = "Add your description here"
+description = "Working with EDA OpenAPI specification"
 readme = "README.md"
-requires-python = ">=3.13"
-dependencies = [
-    "pydantic>=2.10.5",
-]
+requires-python = "~=3.12"
+dependencies = ["httpx==0.28.1", "pydantic==2.10.5"]

src/auth.py

@@ -0,0 +1,81 @@
+import httpx
+
+# Keycloak configuration
+EDA_API_URL = "https://devbox.panda-cobra.ts.net"
+EDA_VERSION = "v24.12.1"
+KC_KEYCLOAK_URL = f"{EDA_API_URL}/core/httpproxy/v1/keycloak/"
+KC_REALM = "master"
+KC_CLIENT_ID = "admin-cli"
+KC_USERNAME = "admin"
+KC_PASSWORD = "admin"
+EDA_REALM = "eda"
+API_CLIENT_ID = "eda"
+
+# Step 1: Get an access token
+token_url = f"{KC_KEYCLOAK_URL}/realms/{KC_REALM}/protocol/openid-connect/token"
+token_data = {
+    "grant_type": "password",
+    "client_id": KC_CLIENT_ID,
+    "username": KC_USERNAME,
+    "password": KC_PASSWORD,
+}
+headers = {"Content-Type": "application/x-www-form-urlencoded"}
+
+
+async def get_client_secret() -> str:
+    async with httpx.AsyncClient(verify=False) as client:
+        # Fetch access token
+        token_response = await client.post(token_url, data=token_data, headers=headers)
+        token_response.raise_for_status()
+        access_token = token_response.json()["access_token"]
+
+        # Step 2: Fetch the `eda` client ID and secret
+        admin_api_url = f"{KC_KEYCLOAK_URL}/admin/realms/{EDA_REALM}/clients"
+        auth_headers = {
+            "Authorization": f"Bearer {access_token}",
+            "Content-Type": "application/json",
+        }
+
+        # Fetch clients
+        clients_response = await client.get(admin_api_url, headers=auth_headers)
+        clients_response.raise_for_status()
+        clients = clients_response.json()
+
+        # Find the `eda` client
+        eda_client = next(
+            (client for client in clients if client["clientId"] == API_CLIENT_ID), None
+        )
+
+        if not eda_client:
+            raise Exception("Client `eda` not found in realm `eda-realm`")
+
+        # Fetch the client secret
+        client_id = eda_client["id"]
+        client_secret_url = f"{admin_api_url}/{client_id}/client-secret"
+        secret_response = await client.get(client_secret_url, headers=auth_headers)
+        secret_response.raise_for_status()
+        client_secret = secret_response.json()["value"]
+
+        return client_secret
+
+
+async def get_access_token(client_secret: str) -> str:
+    token_endpoint = (
+        f"{KC_KEYCLOAK_URL}/realms/{EDA_REALM}/protocol/openid-connect/token"
+    )
+
+    token_data = {
+        "client_id": API_CLIENT_ID,
+        "grant_type": "password",
+        "scope": "openid",
+        "username": "admin",
+        "password": "admin",
+        "client_secret": client_secret,
+    }
+
+    headers = {"Content-Type": "application/x-www-form-urlencoded"}
+
+    async with httpx.AsyncClient(verify=False) as client:
+        response = await client.post(token_endpoint, data=token_data, headers=headers)
+        response.raise_for_status()
+        return response.json()["access_token"]