python: Split generated code into seperate files

python/svix/api/application.py

@@ -0,0 +1,141 @@
+import typing as t
+from dataclasses import asdict, dataclass
+
+
+from ..internal.openapi_client.api.application import (
+    v1_application_create,
+    v1_application_delete,
+    v1_application_get,
+    v1_application_list,
+    v1_application_patch,
+    v1_application_update,
+)
+from ..internal.openapi_client.client import AuthenticatedClient
+from ..internal.openapi_client.models.application_in import ApplicationIn
+from ..internal.openapi_client.models.application_out import ApplicationOut
+from ..internal.openapi_client.models.application_patch import ApplicationPatch
+from ..internal.openapi_client.models.list_response_application_out import (
+    ListResponseApplicationOut,
+)
+from ..internal.openapi_client.models.ordering import Ordering
+
+
+@dataclass
+class ListOptions:
+    iterator: t.Optional[str] = None
+    limit: t.Optional[int] = None
+
+    def to_dict(self) -> t.Dict[str, t.Any]:
+        return {k: v for k, v in asdict(self).items() if v is not None}
+
+
+@dataclass
+class PostOptions:
+    idempotency_key: t.Optional[str] = None
+
+    def to_dict(self) -> t.Dict[str, t.Any]:
+        return {k: v for k, v in asdict(self).items() if v is not None}
+
+
+@dataclass
+class ApplicationListOptions(ListOptions):
+    order: t.Optional[Ordering] = None
+
+
+class ApiBase:
+    _client: AuthenticatedClient
+
+    def __init__(self, client: AuthenticatedClient) -> None:
+        self._client = client
+
+
+class ApplicationAsync(ApiBase):
+    async def list(
+        self, options: ApplicationListOptions = ApplicationListOptions()
+    ) -> ListResponseApplicationOut:
+        return await v1_application_list.request_asyncio(
+            client=self._client, **options.to_dict()
+        )
+
+    async def create(
+        self, application_in: ApplicationIn, options: PostOptions = PostOptions()
+    ) -> ApplicationOut:
+        return await v1_application_create.request_asyncio(
+            client=self._client, json_body=application_in, **options.to_dict()
+        )
+
+    async def get(self, app_id: str) -> ApplicationOut:
+        return await v1_application_get.request_asyncio(
+            client=self._client, app_id=app_id
+        )
+
+    async def get_or_create(
+        self, application_in: ApplicationIn, options: PostOptions = PostOptions()
+    ) -> ApplicationOut:
+        return await v1_application_create.request_asyncio(
+            client=self._client,
+            json_body=application_in,
+            get_if_exists=True,
+            **options.to_dict(),
+        )
+
+    async def update(
+        self, app_id: str, application_in: ApplicationIn
+    ) -> ApplicationOut:
+        return await v1_application_update.request_asyncio(
+            client=self._client, app_id=app_id, json_body=application_in
+        )
+
+    async def patch(
+        self, app_id: str, application_patch: ApplicationPatch
+    ) -> ApplicationOut:
+        return await v1_application_patch.request_asyncio(
+            client=self._client, app_id=app_id, json_body=application_patch
+        )
+
+    async def delete(self, app_id: str) -> None:
+        return await v1_application_delete.request_asyncio(
+            client=self._client, app_id=app_id
+        )
+
+
+class Application(ApiBase):
+    def list(
+        self, options: ApplicationListOptions = ApplicationListOptions()
+    ) -> ListResponseApplicationOut:
+        return v1_application_list.request_sync(
+            client=self._client, **options.to_dict()
+        )
+
+    def create(
+        self, application_in: ApplicationIn, options: PostOptions = PostOptions()
+    ) -> ApplicationOut:
+        return v1_application_create.request_sync(
+            client=self._client, json_body=application_in, **options.to_dict()
+        )
+
+    def get(self, app_id: str) -> ApplicationOut:
+        return v1_application_get.request_sync(client=self._client, app_id=app_id)
+
+    def get_or_create(
+        self, application_in: ApplicationIn, options: PostOptions = PostOptions()
+    ) -> ApplicationOut:
+        return v1_application_create.request_sync(
+            client=self._client,
+            json_body=application_in,
+            get_if_exists=True,
+            **options.to_dict(),
+        )
+
+    def update(self, app_id: str, application_in: ApplicationIn) -> ApplicationOut:
+        return v1_application_update.request_sync(
+            client=self._client, app_id=app_id, json_body=application_in
+        )
+
+    def patch(self, app_id: str, application_patch: ApplicationPatch) -> ApplicationOut:
+        return v1_application_patch.request_sync(
+            client=self._client, app_id=app_id, json_body=application_patch
+        )
+
+    def delete(self, app_id: str) -> None:
+        return v1_application_delete.request_sync(client=self._client, app_id=app_id)

python/svix/api/authentication.py

@@ -0,0 +1,91 @@
+import typing as t
+from dataclasses import asdict, dataclass
+
+
+from ..internal.openapi_client.api.authentication import (
+    v1_authentication_app_portal_access,
+    v1_authentication_dashboard_access,
+    v1_authentication_logout,
+)
+from ..internal.openapi_client.client import AuthenticatedClient
+from ..internal.openapi_client.models.app_portal_access_in import AppPortalAccessIn
+from ..internal.openapi_client.models.app_portal_access_out import AppPortalAccessOut
+from ..internal.openapi_client.models.dashboard_access_out import DashboardAccessOut
+
+
+@dataclass
+class ListOptions:
+    iterator: t.Optional[str] = None
+    limit: t.Optional[int] = None
+
+    def to_dict(self) -> t.Dict[str, t.Any]:
+        return {k: v for k, v in asdict(self).items() if v is not None}
+
+
+@dataclass
+class PostOptions:
+    idempotency_key: t.Optional[str] = None
+
+    def to_dict(self) -> t.Dict[str, t.Any]:
+        return {k: v for k, v in asdict(self).items() if v is not None}
+
+
+class ApiBase:
+    _client: AuthenticatedClient
+
+    def __init__(self, client: AuthenticatedClient) -> None:
+        self._client = client
+
+
+class AuthenticationAsync(ApiBase):
+    async def app_portal_access(
+        self,
+        app_id: str,
+        app_portal_access_in: AppPortalAccessIn,
+        options: PostOptions = PostOptions(),
+    ) -> AppPortalAccessOut:
+        return await v1_authentication_app_portal_access.request_asyncio(
+            client=self._client,
+            app_id=app_id,
+            json_body=app_portal_access_in,
+            **options.to_dict(),
+        )
+
+    async def dashboard_access(
+        self, app_id: str, options: PostOptions = PostOptions()
+    ) -> DashboardAccessOut:
+        return await v1_authentication_dashboard_access.request_asyncio(
+            client=self._client, app_id=app_id, **options.to_dict()
+        )
+
+    async def logout(self, options: PostOptions = PostOptions()) -> None:
+        return await v1_authentication_logout.request_asyncio(
+            client=self._client, **options.to_dict()
+        )
+
+
+class Authentication(ApiBase):
+    def app_portal_access(
+        self,
+        app_id: str,
+        app_portal_access_in: AppPortalAccessIn,
+        options: PostOptions = PostOptions(),
+    ) -> AppPortalAccessOut:
+        return v1_authentication_app_portal_access.request_sync(
+            client=self._client,
+            app_id=app_id,
+            json_body=app_portal_access_in,
+            **options.to_dict(),
+        )
+
+    def dashboard_access(
+        self, app_id: str, options: PostOptions = PostOptions()
+    ) -> DashboardAccessOut:
+        return v1_authentication_dashboard_access.request_sync(
+            client=self._client, app_id=app_id, **options.to_dict()
+        )
+
+    def logout(self, options: PostOptions = PostOptions()) -> None:
+        return v1_authentication_logout.request_sync(
+            client=self._client, **options.to_dict()
+        )

python/svix/api/background_task.py

@@ -0,0 +1,74 @@
+import typing as t
+from dataclasses import asdict, dataclass
+from datetime import datetime, timezone
+
+
+from ..internal.openapi_client.api.background_tasks import (
+    get_background_task,
+    list_background_tasks,
+)
+from ..internal.openapi_client.client import AuthenticatedClient
+from ..internal.openapi_client.models.background_task_out import BackgroundTaskOut
+from ..internal.openapi_client.models.background_task_status import BackgroundTaskStatus
+from ..internal.openapi_client.models.background_task_type import BackgroundTaskType
+from ..internal.openapi_client.models.list_response_background_task_out import (
+    ListResponseBackgroundTaskOut,
+)
+
+DEFAULT_SERVER_URL = "https://api.svix.com"
+
+
+def ensure_tz(x: t.Optional[datetime]) -> t.Optional[datetime]:
+    if x is None:
+        return None
+
+    if x.tzinfo is None:
+        return x.replace(tzinfo=timezone.utc)
+    return x
+
+
+@dataclass
+class ListOptions:
+    iterator: t.Optional[str] = None
+    limit: t.Optional[int] = None
+
+    def to_dict(self) -> t.Dict[str, t.Any]:
+        return {k: v for k, v in asdict(self).items() if v is not None}
+
+
+class BackgroundTaskListOptions(ListOptions):
+    status: t.Optional[BackgroundTaskStatus] = None
+    task: t.Optional[BackgroundTaskType] = None
+
+
+class ApiBase:
+    _client: AuthenticatedClient
+
+    def __init__(self, client: AuthenticatedClient) -> None:
+        self._client = client
+
+
+class BackgroundTaskAsync(ApiBase):
+    async def list(
+        self, options: BackgroundTaskListOptions = BackgroundTaskListOptions()
+    ) -> ListResponseBackgroundTaskOut:
+        return await list_background_tasks.request_asyncio(
+            client=self._client, **options.to_dict()
+        )
+
+    async def get(self, task_id: str) -> BackgroundTaskOut:
+        return await get_background_task.request_asyncio(
+            client=self._client, task_id=task_id
+        )
+
+
+class BackgroundTask(ApiBase):
+    def list(
+        self, options: BackgroundTaskListOptions = BackgroundTaskListOptions()
+    ) -> ListResponseBackgroundTaskOut:
+        return list_background_tasks.request_sync(
+            client=self._client, **options.to_dict()
+        )
+
+    def get(self, task_id: str) -> BackgroundTaskOut:
+        return get_background_task.request_sync(client=self._client, task_id=task_id)