SRTP-769-implement-api-send-gpd-message

Author: LarissaASLeite

.github/workflows/deploy_with_github_runner.yml

@@ -0,0 +1,34 @@
+name: Deploy on AKS
+
+on:
+  push:
+    branches:
+      - '**'
+  workflow_dispatch:
+    inputs:
+      environment:
+        required: true
+        type: string
+        default: dev
+        description: env
+
+env:
+  NAMESPACE: afm
+  APP_NAME: rtp-platform-qa
+
+permissions:
+  id-token: write
+  contents: read
+
+jobs:
+  deploy:
+    runs-on: [ self-hosted, self-hosted-job, "${{ github.event.inputs.environment || 'dev' }}" ]
+    environment: ${{ github.event.inputs.environment || 'dev' }}
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Check runner environment
+        run: |
+          echo "🏃 Running on environment: ${{ github.event.inputs.environment || 'dev' }}"
+          echo "📦 Namespace: ${{ env.NAMESPACE }}"
+          echo "📦 App: ${{ env.APP_NAME }}"

.gitignore

@@ -122,7 +122,7 @@ celerybeat.pid
 *.sage.py
 
 # Environments
-.env
+gpd-test/.env
 .venv
 env/
 venv/

gpd-test/dto.py

@@ -0,0 +1,29 @@
+from enum import Enum
+from pydantic import BaseModel, Field
+
+class RTPOperationCode(str, Enum):
+  CREATE = "CREATE"
+  UPDATE = "UPDATE"
+  DELETE = "DELETE"
+
+class PaymentPositionStatus(str, Enum):
+  CREATED = "CREATED"
+  SENT = "SENT"
+  ACCEPTED = "ACCEPTED"
+
+class RTPMessage(BaseModel):
+  id: int
+  operation: RTPOperationCode
+  timestamp: int
+  iuv: str
+  subject: str
+  description: str
+  ec_tax_code: str = Field(..., alias="ec_tax_code")
+  debtor_tax_code: str = Field(..., alias="debtor_tax_code")
+  nav: str
+  due_date: int = Field(..., alias="due_date")
+  amount: int
+  status: PaymentPositionStatus
+  psp_code: str = Field(..., alias="psp_code")
+  psp_tax_code: str = Field(..., alias="psp_tax_code")
+  is_partial_payment: bool = Field(..., alias="is_partial_payment")

gpd-test/keyvault.py

@@ -0,0 +1,31 @@
+# from azure.identity import AzureCliCredential
+# from azure.keyvault.secrets import SecretClient
+# import os
+#
+# def get_eventhub_connection_string():
+#     keyvault_name = os.environ["KEYVAULT_NAME"]
+#     secret_name = os.environ["EVENTHUB_SECRET_NAME"]
+#
+#     keyvault_url = f"https://{keyvault_name}.vault.azure.net"
+#     credential = AzureCliCredential()
+#     client = SecretClient(vault_url=keyvault_url, credential=credential)
+#
+#     secret = client.get_secret(secret_name)
+#     print(f"[KeyVault] Retrieved secret '{secret_name}' from '{keyvault_name}'")
+#     return secret.value
+
+from azure.identity import DefaultAzureCredential
+from azure.keyvault.secrets import SecretClient
+import os
+
+def get_eventhub_connection_string():
+    keyvault_name = os.environ["KEYVAULT_NAME"]
+    secret_name = os.environ["EVENTHUB_SECRET_NAME"]
+
+    keyvault_url = f"https://{keyvault_name}.vault.azure.net"
+    credential = DefaultAzureCredential()
+    client = SecretClient(vault_url=keyvault_url, credential=credential)
+
+    secret = client.get_secret(secret_name)
+    print(f"[KeyVault] Retrieved secret '{secret_name}' from '{keyvault_name}'")
+    return secret.value

gpd-test/main.py

@@ -0,0 +1,31 @@
+from fastapi import FastAPI, HTTPException, Request
+from contextlib import asynccontextmanager
+from producer import setup_producer
+from dto import RTPMessage
+from dotenv import load_dotenv
+import os, json
+
+load_dotenv()
+
+EVENTHUB_TOPIC = os.environ["EVENTHUB_TOPIC"]
+
+@asynccontextmanager
+async def app_lifespan(fastapi_app: FastAPI):
+    kafka_producer = await setup_producer()
+    app.state.producer = kafka_producer
+    yield
+    await kafka_producer.stop()
+
+app = FastAPI(lifespan=app_lifespan)
+
+@app.post("/send/gpd/message")
+async def send_msg(message: RTPMessage, request: Request):
+    try:
+        producer = request.app.state.producer
+        await producer.send_and_wait(
+            EVENTHUB_TOPIC,
+            json.dumps(message.model_dump(by_alias=True)).encode("utf-8")
+        )
+        return {"status": "ok"}
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))

gpd-test/producer.py

@@ -0,0 +1,23 @@
+from aiokafka import AIOKafkaProducer
+import os, ssl
+from keyvault import get_eventhub_connection_string
+
+async def setup_producer():
+    connection_string = get_eventhub_connection_string()
+    namespace = os.environ["EVENTHUB_NAMESPACE"]
+
+    ssl_context = ssl.create_default_context()
+    ssl_context.check_hostname = True
+    ssl_context.verify_mode = ssl.CERT_REQUIRED
+
+    producer = AIOKafkaProducer(
+        bootstrap_servers=f"{namespace}.servicebus.windows.net:9093",
+        security_protocol="SASL_SSL",
+        sasl_mechanism="PLAIN",
+        sasl_plain_username="$ConnectionString",
+        sasl_plain_password=connection_string,
+        ssl_context=ssl_context,
+        client_id="python-producer"
+    )
+    await producer.start()
+    return producer

requirements.txt

@@ -7,3 +7,11 @@ faker==22.5.1
 requests==2.32.4
 schwifty==2025.1.0
 urllib3==2.5.0
+
+python-dotenv~=1.1.1
+pydantic~=2.11.7
+pytest~=8.4.1
+fastapi~=0.116.1
+aiokafka~=0.12.0
+keyvault~=0.2.1
+azure-core~=1.35.0