✨ Feature: introduce api versioning, lazy loading and webhook namespa…

…ce (#73)

codegen/config.py

@@ -1,26 +1,53 @@
-from typing import Any, Dict, List
+from typing import Any
+from pathlib import Path
 
 from pydantic import Field, BaseModel
 
 
-class Overridable(BaseModel):
-    class_overrides: Dict[str, str] = Field(default_factory=dict)
-    field_overrides: Dict[str, str] = Field(default_factory=dict)
-    schema_overrides: Dict[str, Dict[str, Any]] = Field(default_factory=dict)
+class Override(BaseModel):
+    class_overrides: dict[str, str] = Field(default_factory=dict)
+    field_overrides: dict[str, str] = Field(default_factory=dict)
+    schema_overrides: dict[str, dict[str, Any]] = Field(default_factory=dict)
 
 
-class RestConfig(Overridable):
-    version: str
-    description_source: str
-    output_dir: str
-
+class VersionedOverride(Override):
+    target_versions: list[str] = Field(default_factory=list)
 
-class WebhookConfig(Overridable):
-    schema_source: str
-    output: str
-    types_output: str
 
-
-class Config(Overridable):
-    rest: List[RestConfig]
-    webhook: WebhookConfig
+class DescriptionConfig(BaseModel):
+    version: str
+    is_latest: bool = False
+    """If true, the description will be used as the default description."""
+    source: str
+
+
+class Config(BaseModel):
+    output_dir: Path
+    legacy_rest_models: Path
+    version_prefix: str = "v"
+    descriptions: list[DescriptionConfig]
+    overrides: list[VersionedOverride] = Field(default_factory=list)
+
+    def get_override_config_for_version(self, version: str) -> Override:
+        selected_overrides = [
+            override
+            for override in self.overrides
+            if version in override.target_versions or not override.target_versions
+        ]
+        return Override(
+            class_overrides={
+                key: value
+                for override in selected_overrides
+                for key, value in override.class_overrides.items()
+            },
+            field_overrides={
+                key: value
+                for override in selected_overrides
+                for key, value in override.field_overrides.items()
+            },
+            schema_overrides={
+                key: value
+                for override in selected_overrides
+                for key, value in override.schema_overrides.items()
+            },
+        )

codegen/parser/__init__.py

@@ -1,19 +1,23 @@
 from contextvars import ContextVar
-from typing import Dict, List, Tuple, Union, Optional
+from typing import TYPE_CHECKING, Optional
 
 import httpx
 from openapi_pydantic import OpenAPI
 
+if TYPE_CHECKING:
+    from ..source import Source
+    from ..config import Override
+
 # parser context
-_override_config: ContextVar[Tuple["Overridable", ...]] = ContextVar("override_config")
-_schemas: ContextVar[Dict[httpx.URL, "SchemaData"]] = ContextVar("schemas")
+_override_config: ContextVar["Override"] = ContextVar("override_config")
+_schemas: ContextVar[dict[httpx.URL, "SchemaData"]] = ContextVar("schemas")
 
 
-def get_override_config() -> Tuple["Overridable", ...]:
+def get_override_config() -> "Override":
     return _override_config.get()
 
 
-def get_schemas() -> Dict[httpx.URL, "SchemaData"]:
+def get_schemas() -> dict[httpx.URL, "SchemaData"]:
     return _schemas.get()
 
 
@@ -27,102 +31,62 @@ def add_schema(ref: httpx.URL, schema: "SchemaData"):
     _schemas.get()[ref] = schema
 
 
-from ..source import Source
 from .utils import merge_dict
+from .webhooks import parse_webhook
 from .endpoints import parse_endpoint
 from .utils import sanitize as sanitize
 from .utils import kebab_case as kebab_case
 from .utils import snake_case as snake_case
 from .data import OpenAPIData as OpenAPIData
 from .data import WebhookData as WebhookData
 from .utils import pascal_case as pascal_case
-from .endpoints import EndpointData as EndpointData
-from .schemas import SchemaData, UnionSchema, parse_schema
+from .data import EndpointData as EndpointData
+from .schemas import SchemaData, ModelSchema, parse_schema
 from .utils import fix_reserved_words as fix_reserved_words
-from ..config import Config, RestConfig, Overridable, WebhookConfig
 
 
-def parse_openapi_spec(source: Source, rest: RestConfig, config: Config) -> OpenAPIData:
+def parse_openapi_spec(source: "Source", override: "Override") -> OpenAPIData:
     source = source.get_root()
 
-    # apply schema overrides first
-    for path, new_schema in {
-        **config.schema_overrides,
-        **rest.schema_overrides,
-    }.items():
+    # apply schema overrides first to make sure json pointer is correct
+    for path, new_schema in override.schema_overrides.items():
         ref = str(httpx.URL(fragment=path))
         merge_dict(source.resolve_ref(ref).data, new_schema)
 
-    _ot = _override_config.set((rest, config))
+    _ot = _override_config.set(override)
     _st = _schemas.set({})
 
     try:
         openapi = OpenAPI.model_validate(source.root)
 
-        # cache /components/schemas first
+        # pre-cache /components/schemas first
         if openapi.components and openapi.components.schemas:
             schemas_source = source / "components" / "schemas"
             for name in openapi.components.schemas:
                 schema_source = schemas_source / name
                 parse_schema(schema_source, name)
 
-        endpoints: List[EndpointData] = []
+        # load endpoints
+        endpoints: list[EndpointData] = []
         if openapi.paths:
             for path in openapi.paths:
                 endpoints.extend(parse_endpoint(source / "paths" / path, path))
 
+        # load webhooks
+        webhooks: list[WebhookData] = []
+        if openapi.webhooks:
+            for webhook in openapi.webhooks:
+                if webhook_data := parse_webhook(source / "webhooks" / webhook):
+                    webhooks.append(webhook_data)
+
         return OpenAPIData(
-            title=openapi.info.title,
-            description=openapi.info.description,
-            version=openapi.info.version,
+            models=[
+                schema
+                for schema in get_schemas().values()
+                if isinstance(schema, ModelSchema)
+            ],
             endpoints=endpoints,
-            schemas=list(get_schemas().values()),
-        )
-    finally:
-        _override_config.reset(_ot)
-        _schemas.reset(_st)
-
-
-def parse_webhook_schema(
-    source: Source, webhook: WebhookConfig, config: Config
-) -> WebhookData:
-    source = source.get_root()
-
-    # apply schema overrides first
-    for path, new_schema in {
-        **config.schema_overrides,
-        **webhook.schema_overrides,
-    }.items():
-        ref = str(httpx.URL(fragment=path))
-        merge_dict(source.resolve_ref(ref).data, new_schema)
-
-    _ot = _override_config.set((webhook, config))
-    _st = _schemas.set({})
-
-    try:
-        root_schema = parse_schema(source, "webhook_schema")
-        if not isinstance(root_schema, UnionSchema):
-            raise TypeError("Webhook root schema must be a UnionSchema")
-
-        schemas = get_schemas()
-        definitions: Dict[str, Union[SchemaData, Dict[str, SchemaData]]] = {}
-        for event in source.data["oneOf"]:
-            event_name = event["$ref"].split("/")[-1]
-            event_source = source.resolve_ref(event["$ref"])
-            schema = schemas[event_source.uri]
-            if isinstance(schema, UnionSchema):
-                definitions[event_name] = {
-                    action["$ref"].split("/")[-1]: schemas[
-                        event_source.resolve_ref(action["$ref"]).uri
-                    ]
-                    for action in event_source.data["oneOf"]
-                }
-            else:
-                definitions[event_name] = schema
-
-        return WebhookData(
-            schemas=list(schemas.values()),
-            definitions=definitions,
+            webhooks=webhooks,
         )
     finally:
         _override_config.reset(_ot)