feat: also generate websocket models (#5)

still missing good tests :-(
#8

README.md

@@ -38,6 +38,18 @@ my_json = {
 my_strongly_typed_record = ApiRecord.model_validate(my_json)
 ```
 
+or
+
+```python
+from verzeichnisdienst.v1.websocket import Contact
+
+my_json_contact = {
+    "email": "verzeichnis@hochfrequenz.de",
+    "phone": "0049123457890"
+}
+my_strongly_typed_contact = Contact.model_validate(my_json_contact)
+```
+
 See the [tests](unittests/test_models.py) for more examples.
 
 ## Project Structure

domain-specific-terms.txt

@@ -3,3 +3,6 @@ adresse
 oder
 feld
 ist
+signatur
+felds
+als

pyproject.toml

@@ -76,7 +76,7 @@ truethy-bool = true
 disable_error_code = []
 
 [[tool.mypy.overrides]]
-module = "verzeichnisdienst.v1._autogenerated"
+module = ["verzeichnisdienst.v1._autogenerated", "verzeichnisdienst.v1._autogenerated_websocket"]
 ignore_errors = true
 
 

src/verzeichnisdienst/__init__.py

@@ -1,5 +1,6 @@
 """
-The package verzeichnisdienst contains model classes that are used in the Verzeichnisdienst API by EDI@Energy.
+The package verzeichnisdienst contains model classes that are used in the Verzeichnisdienst API
+by EDI@Energy.
 The raw data are published on Swagger Hub:
 https://app.swaggerhub.com/apis-docs/edi_energy/Verzeichnisdienst_Web-API_2024_10_01/1.0.0
 """

src/verzeichnisdienst/v1/_autogenerated.py

@@ -1,6 +1,6 @@
 # generated by datamodel-codegen:
 #   filename:  openapi.yml
-#   timestamp: 2024-11-18T08:44:34+00:00
+#   timestamp: 2024-11-18T09:36:57+00:00
 
 from __future__ import annotations
 

src/verzeichnisdienst/v1/_autogenerated_websocket.py

@@ -0,0 +1,188 @@
+# generated by datamodel-codegen:
+#   filename:  openapi_websocket.yml
+#   timestamp: 2024-11-18T09:36:59+00:00
+
+from __future__ import annotations
+
+from enum import Enum
+from typing import Dict, List, Optional
+
+from pydantic import BaseModel, ConfigDict, Field, conint
+
+
+class Status(Enum):
+    Offline = "Offline"
+    Test = "Test"
+    Maintenance = "Maintenance"
+    Online = "Online"
+
+
+class ApiRecord(BaseModel):
+    model_config = ConfigDict(
+        extra="forbid",
+    )
+    providerId: str = Field(
+        ...,
+        description="Eindeutige Kennung des verantwortlichen API-Anbieters; Hinweise in Kapitel [Einträge im Verzeichnis](#einträge-im-verzeichnis) beachten.",
+    )
+    apiId: str = Field(
+        ...,
+        description="Eindeutige Kennung des API-Webdiensts; Hinweise in Kapitel [Einträge im Verzeichnis](#einträge-im-verzeichnis) beachten.",
+    )
+    majorVersion: int = Field(
+        ...,
+        description="Major-Version des API-Webdiensts; Hinweise in Kapitel [Einträge im Verzeichnis](#einträge-im-verzeichnis) beachten.",
+    )
+    url: str = Field(..., description="Adresse des Endpunkts, an dem der API-Webdienst aufgerufen werden kann.")
+    additionalMetadata: Optional[Dict[str, str]] = Field(
+        None,
+        description="Zusätzliche Informationen über den Aufruf oder die Antworten des entsprechenden API-Webdiensts in Form von Schlüssel-Wert-Paaren. Dieses Feld ist optional und es wird durch übergreifende Regelungen außerhalb dieser Spezifikation sowie durch die Spezifikation des entsprechenden API-Webdiensts festgelegt, welche Informationen zu hinterlegen sind.",
+    )
+    lastUpdated: str = Field(..., description="Zeitpunkt der letzten Aktualisierung des Verzeichniseintrags.")
+    revision: conint(ge=1) = Field(
+        ...,
+        description="Fortlaufende Revisionsnummer des Verzeichniseintrags. Wird bei jeder Aktualisierung des Eintrags inkrementiert.",
+    )
+    status: Status = Field(
+        ...,
+        description="Aktueller Status des hinterlegten API-Endpunkts; Hinweise in Kapitel [Statusmodell](#statusmodell) beachten.",
+    )
+
+
+class ApiRecordRef(BaseModel):
+    model_config = ConfigDict(
+        extra="forbid",
+    )
+    providerId: str = Field(
+        ...,
+        description="Eindeutige Kennung des verantwortlichen API-Anbieters; Hinweise in Kapitel [Einträge im Verzeichnis](#einträge-im-verzeichnis) beachten.",
+    )
+    apiId: str = Field(
+        ...,
+        description="Eindeutige Kennung des API-Webdiensts; Hinweise in Kapitel [Einträge im Verzeichnis](#einträge-im-verzeichnis) beachten.",
+    )
+    majorVersion: int = Field(
+        ...,
+        description="Major Version des API-Webdiensts; Hinweise in Kapitel [Einträge im Verzeichnis](#einträge-im-verzeichnis) beachten.",
+    )
+
+
+class RedirectedItem(BaseModel):
+    model_config = ConfigDict(
+        extra="forbid",
+    )
+    recordRef: ApiRecordRef
+    url: Optional[str] = Field(
+        None, description="Konfigurierte Ziel-URL des Redirects. Ist nicht vorhanden, wenn der Redirect entfernt wurde."
+    )
+
+
+class CanceledItem(BaseModel):
+    model_config = ConfigDict(
+        extra="forbid",
+    )
+    recordRef: ApiRecordRef
+    canceledByClient: bool = Field(
+        ...,
+        description="Zeigt an, ob die Subscription durch den Client (true) oder durch den Server (false) beendet wurde.",
+    )
+    reason: Optional[str] = Field(None, description="Erläuternde Beschreibung, weshalb die Subscription beendet wurde.")
+
+
+class Error(BaseModel):
+    model_config = ConfigDict(
+        extra="forbid",
+    )
+    statusCode: conint(ge=0) = Field(..., description="HTTP Status Code, der den Fehler beschreibt.")
+    description: str = Field(..., description="Informativer Text, der den Fehler beschreibt.")
+    request: Optional[str] = Field(
+        None, description="Base64-kodierter SubscriptionRequest, der den Fehler ausgelöst hat."
+    )
+
+
+class Contact(BaseModel):
+    model_config = ConfigDict(
+        extra="forbid",
+    )
+    email: Optional[str] = Field(None, description="E-Mail-Adresse des technischen Supports.")
+    phone: Optional[str] = Field(None, description="Telefonnummer des technischen Supports.")
+
+
+class ServiceInfo(BaseModel):
+    model_config = ConfigDict(
+        extra="forbid",
+    )
+    version: str = Field(
+        ..., description="Vollqualifizierte Versionsnummer der implementierten Schnittstelle des Verzeichnisdiensts."
+    )
+    contact: Contact = Field(
+        ...,
+        description="Kontaktdaten des technischen Betreibers.\n\n*__Hinweis:__ Die einzelnen Felder sind OPTIONAL, jedoch MUSS mindestens eins befüllt sein.*",
+    )
+    lastUpdated: str = Field(..., description="Zeitpunkt der letzten Aktualisierung dieser Information.")
+    revision: conint(ge=1) = Field(
+        ...,
+        description="Fortlaufende Revisionsnummer dieser Information. Beginnt bei 1 und wird bei jeder Aktualisierung inkrementiert.",
+    )
+
+
+class SignedApiRecord(BaseModel):
+    model_config = ConfigDict(
+        extra="forbid",
+    )
+    content: ApiRecord
+    signature: str = Field(
+        ...,
+        description="Die Signatur des Verzeichniseintrags; Hinweise in Kapitel [Signatur von Verzeichniseinträgen](#signatur-von-verzeichniseinträgen) beachten.",
+    )
+    signingCert: str = Field(
+        ...,
+        description="Das verwendete Signaturzertifikat; Hinweise in Kapitel [Signatur von Verzeichniseinträgen](#signatur-von-verzeichniseinträgen) beachten.",
+    )
+
+
+class RequestedItem(BaseModel):
+    model_config = ConfigDict(
+        extra="forbid",
+    )
+    recordRef: ApiRecordRef
+    knownRevision: Optional[conint(ge=0)] = Field(
+        None,
+        description="Beim Client bekannte Revision des angeforderten Verzeichniseintrags. Der Wert 0 oder die Abwesenheit dieses Felds bedeutet, dass dem Client kein Eintrag bekannt ist.",
+    )
+
+
+class SubscriptionRequest(BaseModel):
+    model_config = ConfigDict(
+        extra="forbid",
+    )
+    id: str = Field(
+        ...,
+        description="ID des Subscription Requests. Wird durch den Client selbst gewählt und innerhalb des Verzeichnisdiensts nicht ausgewertet. Dieser Wert wird zur Korrelation vom Request in die dazugehörige [DirectoryNotification](#schema-DirectoryNotification) übernommen.",
+    )
+    requested: Optional[List[RequestedItem]] = Field(None, description="Zu erstellende Subscriptions.")
+    canceled: Optional[List[ApiRecordRef]] = Field(None, description="Zu kündigende Subscriptions.")
+
+
+class DirectoryNotification(BaseModel):
+    model_config = ConfigDict(
+        extra="forbid",
+    )
+    subscriptionId: Optional[str] = Field(
+        None,
+        description="ID des [SubscriptionRequest](#schema-SubscriptionRequest)-Objekts, auf den sich die Benachrichtigung bezieht.",
+    )
+    timestamp: str = Field(
+        ...,
+        description="Der Zeitpunkt, an dem diese Benachrichtigung erzeugt wurde. Der Wert MUSS gemäß ISO 8601 formatiert und als UTC angeben werden.",
+    )
+    serviceInfo: Optional[ServiceInfo] = None
+    modified: Optional[List[SignedApiRecord]] = Field(
+        None, description="Hinzugefügte oder geänderte Verzeichniseinträge."
+    )
+    redirected: Optional[List[RedirectedItem]] = Field(
+        None, description="Konfigurierte Redirects für Verzeichniseinträge."
+    )
+    deleted: Optional[List[ApiRecordRef]] = Field(None, description="Gelöschte Verzeichniseinträge.")
+    canceled: Optional[List[CanceledItem]] = Field(None, description="Beendete Subscriptions.")
+    error: Optional[Error] = None

src/verzeichnisdienst/v1/websocket.py

@@ -0,0 +1,36 @@
+"""
+model classes for EDI@Energy Verzeichnisdienst Websocket API
+"""
+
+# In this module we just re-export the _autogenerated_websocket classes for now.
+# But we might also fix problems in the openapi_websocket.yml here.
+
+from ._autogenerated_websocket import (
+    ApiRecord,
+    ApiRecordRef,
+    CanceledItem,
+    Contact,
+    DirectoryNotification,
+    Error,
+    RedirectedItem,
+    RequestedItem,
+    ServiceInfo,
+    SignedApiRecord,
+    Status,
+    SubscriptionRequest,
+)
+
+__all__ = [
+    "Status",
+    "ApiRecord",
+    "ApiRecordRef",
+    "SignedApiRecord",
+    "ServiceInfo",
+    "SubscriptionRequest",
+    "RedirectedItem",
+    "RequestedItem",
+    "DirectoryNotification",
+    "CanceledItem",
+    "Error",
+    "Contact",
+]

tox.ini

@@ -26,7 +26,7 @@ deps =
     # add your fixtures like e.g. pytest_datafiles here
 setenv = PYTHONPATH = {toxinidir}/src
 commands =
-    pylint --ignore=_autogenerated.py verzeichnisdienst
+    pylint --ignore=_autogenerated.py,_autogenerated_websocket.py verzeichnisdienst
     pylint unittests --rcfile=unittests/.pylintrc
     # add single files (ending with .py) or packages here
 
@@ -78,6 +78,9 @@ commands =
     datamodel-codegen --encoding utf-8 --input openapi/v1/openapi.yml --output src/verzeichnisdienst/v1/_autogenerated.py --input-file-type openapi --output-model-type pydantic_v2.BaseModel --output-datetime-class AwareDatetime
     black src/verzeichnisdienst/v1/_autogenerated.py
     isort src/verzeichnisdienst/v1/_autogenerated.py
+    datamodel-codegen --encoding utf-8 --input openapi/v1/openapi_websocket.yml --output src/verzeichnisdienst/v1/_autogenerated_websocket.py --input-file-type openapi --output-model-type pydantic_v2.BaseModel --output-datetime-class AwareDatetime
+    black src/verzeichnisdienst/v1/_autogenerated_websocket.py
+    isort src/verzeichnisdienst/v1/_autogenerated_websocket.py
 
 [testenv:dev]
 # the dev environment contains everything you need to start developing on your local machine.

unittests/example_ws_jsons_v1.py

@@ -0,0 +1,19 @@
+api_record = {  # copied from swagger hub
+    "providerId": "1234567890123",
+    "apiId": "example",
+    "majorVersion": 1,
+    "url": "https://www.example.org/api/resource/v1",
+    "additionalMetadata": None,
+    "lastUpdated": "2024-10-01T00:00:00Z",  # changed this to Z because pydantic prefers 'Z' over '+00:00'
+    "revision": 1,
+    "status": "Test",
+}
+
+contact_info = {"email": "foo@bar.inv", "phone": "00491234567890"}  # handwritten
+
+service_info = {  # handwritten
+    "version": "v1.2.3",
+    "contact": contact_info,
+    "lastUpdated": "2024-11-16T00:00:00Z",
+    "revision": 17,
+}

unittests/test_ws_models_v1.py

@@ -0,0 +1,19 @@
+from typing import Any
+
+import pytest
+
+from verzeichnisdienst.v1.websocket import ApiRecord
+
+from .example_ws_jsons_v1 import api_record
+
+
+@pytest.mark.parametrize(
+    "json_dict",
+    [
+        api_record,
+    ],
+)
+def test_api_record(json_dict: dict[str, Any]) -> None:
+    model = ApiRecord.model_validate(json_dict)
+    re_serialized = model.model_dump(mode="json")
+    assert re_serialized == json_dict