Skip to content

Commit

Permalink
python: Add openapi-codegen generated integration.py
Browse files Browse the repository at this point in the history
Breaking change: in the create and update function the integ_(in/update) is renamed to integration_(in/update)

This is due to the way that the codegen decides how to name function arg. (The type of the function arg to_snake_case)
I don't think it makes sense to add a special case in the codegen to shorten `integration` to `integ`
This problem is a bit more annoying, since the path param is called `integ_id` (in the `openapi.json`). So to keep consistency we may want to add the special case to the codegen

Alternatively (and this would make me happy) we could change the path param name from `integ_id` to `integration_id`. (in the `openapi.json`) This will make the file consistent, and won't require a special case in the openapi-codegen
  • Loading branch information
svix-mman committed Jan 16, 2025
1 parent 2fb1cf2 commit 0a45271
Showing 1 changed file with 71 additions and 46 deletions.
117 changes: 71 additions & 46 deletions python/svix/api/integration.py
Original file line number Diff line number Diff line change
@@ -1,72 +1,93 @@
import typing as t
from dataclasses import dataclass

from .common import ApiBase, BaseOptions
from ..internal.openapi_client import models


from ..internal.openapi_client.api.integration import (
v1_integration_list,
v1_integration_create,
v1_integration_delete,
v1_integration_get,
v1_integration_update,
v1_integration_delete,
v1_integration_get_key,
v1_integration_list,
v1_integration_rotate_key,
v1_integration_update,
)
from ..internal.openapi_client.models.integration_in import IntegrationIn
from ..internal.openapi_client.models.integration_key_out import IntegrationKeyOut
from ..internal.openapi_client.models.integration_out import IntegrationOut
from ..internal.openapi_client.models.integration_update import IntegrationUpdate

from ..internal.openapi_client.models.list_response_integration_out import (
ListResponseIntegrationOut,
)
from .common import ListOptions, PostOptions, ApiBase
from ..internal.openapi_client.models.integration_in import IntegrationIn
from ..internal.openapi_client.models.integration_out import IntegrationOut
from ..internal.openapi_client.models.integration_update import IntegrationUpdate
from ..internal.openapi_client.models.integration_key_out import IntegrationKeyOut


@dataclass
class IntegrationListOptions(BaseOptions):
# Limit the number of returned items
limit: t.Optional[int] = None
# The iterator returned from a prior invocation
iterator: t.Optional[str] = None
# The sorting order of the returned items
order: t.Optional[models.Ordering] = None


@dataclass
class IntegrationCreateOptions(BaseOptions):
idempotency_key: t.Optional[str] = None


@dataclass
class IntegrationListOptions(ListOptions):
pass
class IntegrationRotateKeyOptions(BaseOptions):
idempotency_key: t.Optional[str] = None


class IntegrationAsync(ApiBase):
async def list(
self, app_id: str, options: IntegrationListOptions = IntegrationListOptions()
) -> ListResponseIntegrationOut:
"""List the application's integrations."""
return await v1_integration_list.request_asyncio(
client=self._client,
app_id=app_id,
**options.to_dict(),
client=self._client, app_id=app_id, **options.to_dict()
)

async def create(
self, app_id: str, integ_in: IntegrationIn, options: PostOptions = PostOptions()
self,
app_id: str,
integration_in: IntegrationIn,
options: IntegrationCreateOptions = IntegrationCreateOptions(),
) -> IntegrationOut:
"""Create an integration."""
return await v1_integration_create.request_asyncio(
client=self._client,
app_id=app_id,
json_body=integ_in,
json_body=integration_in,
**options.to_dict(),
)

async def get(self, app_id: str, integ_id: str) -> IntegrationOut:
"""Get an integration."""
return await v1_integration_get.request_asyncio(
client=self._client,
app_id=app_id,
integ_id=integ_id,
client=self._client, app_id=app_id, integ_id=integ_id
)

async def update(
self, app_id: str, integ_id: str, integ_update: IntegrationUpdate
self, app_id: str, integ_id: str, integration_update: IntegrationUpdate
) -> IntegrationOut:
"""Update an integration."""
return await v1_integration_update.request_asyncio(
client=self._client,
app_id=app_id,
integ_id=integ_id,
json_body=integ_update,
json_body=integration_update,
)

async def delete(self, app_id: str, integ_id: str) -> None:
"""Delete an integration."""
return await v1_integration_delete.request_asyncio(
client=self._client,
app_id=app_id,
integ_id=integ_id,
client=self._client, app_id=app_id, integ_id=integ_id
)

async def get_key(self, app_id: str, integ_id: str) -> IntegrationKeyOut:
Expand All @@ -77,58 +98,61 @@ async def get_key(self, app_id: str, integ_id: str) -> IntegrationKeyOut:
)

async def rotate_key(
self, app_id: str, integ_id: str, options: PostOptions = PostOptions()
self,
app_id: str,
integ_id: str,
options: IntegrationRotateKeyOptions = IntegrationRotateKeyOptions(),
) -> IntegrationKeyOut:
"""Rotate the integration's key. The previous key will be immediately revoked."""
return await v1_integration_rotate_key.request_asyncio(
client=self._client,
app_id=app_id,
integ_id=integ_id,
**options.to_dict(),
client=self._client, app_id=app_id, integ_id=integ_id, **options.to_dict()
)


class Integration(ApiBase):
def list(
self, app_id: str, options: IntegrationListOptions = IntegrationListOptions()
) -> ListResponseIntegrationOut:
"""List the application's integrations."""
return v1_integration_list.request_sync(
client=self._client,
app_id=app_id,
**options.to_dict(),
client=self._client, app_id=app_id, **options.to_dict()
)

def create(
self, app_id: str, integ_in: IntegrationIn, options: PostOptions = PostOptions()
self,
app_id: str,
integration_in: IntegrationIn,
options: IntegrationCreateOptions = IntegrationCreateOptions(),
) -> IntegrationOut:
"""Create an integration."""
return v1_integration_create.request_sync(
client=self._client,
app_id=app_id,
json_body=integ_in,
json_body=integration_in,
**options.to_dict(),
)

def get(self, app_id: str, integ_id: str) -> IntegrationOut:
"""Get an integration."""
return v1_integration_get.request_sync(
client=self._client,
app_id=app_id,
integ_id=integ_id,
client=self._client, app_id=app_id, integ_id=integ_id
)

def update(
self, app_id: str, integ_id: str, integ_update: IntegrationUpdate
self, app_id: str, integ_id: str, integration_update: IntegrationUpdate
) -> IntegrationOut:
"""Update an integration."""
return v1_integration_update.request_sync(
client=self._client,
app_id=app_id,
integ_id=integ_id,
json_body=integ_update,
json_body=integration_update,
)

def delete(self, app_id: str, integ_id: str) -> None:
"""Delete an integration."""
return v1_integration_delete.request_sync(
client=self._client,
app_id=app_id,
integ_id=integ_id,
client=self._client, app_id=app_id, integ_id=integ_id
)

def get_key(self, app_id: str, integ_id: str) -> IntegrationKeyOut:
Expand All @@ -139,11 +163,12 @@ def get_key(self, app_id: str, integ_id: str) -> IntegrationKeyOut:
)

def rotate_key(
self, app_id: str, integ_id: str, options: PostOptions = PostOptions()
self,
app_id: str,
integ_id: str,
options: IntegrationRotateKeyOptions = IntegrationRotateKeyOptions(),
) -> IntegrationKeyOut:
"""Rotate the integration's key. The previous key will be immediately revoked."""
return v1_integration_rotate_key.request_sync(
client=self._client,
app_id=app_id,
integ_id=integ_id,
**options.to_dict(),
client=self._client, app_id=app_id, integ_id=integ_id, **options.to_dict()
)

0 comments on commit 0a45271

Please sign in to comment.