Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: Improve typehints #379

Merged
merged 2 commits into from
May 31, 2024
Merged
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 11 additions & 5 deletions sdks/python/julep/managers/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,14 @@
from functools import wraps
from typing import Callable
from uuid import UUID
from typing_extensions import ParamSpec
from pydantic.main import Model
from ..api.types import ResourceCreatedResponse


P = ParamSpec("P")


class NotSet:
pass

Expand Down Expand Up @@ -33,19 +38,20 @@ def is_valid_uuid4(uuid_to_test: str) -> bool:
return True


def rewrap_in_class(cls):
def decorator(func: Callable[..., ResourceCreatedResponse]):
@wraps(func)
def rewrap_in_class(cls: type[Model]):
def decorator(func: Callable[P, ResourceCreatedResponse]):
# This wrapper is used for asynchronous functions to ensure they are properly awaited and their results are processed by `cls.construct`.
@wraps(func)
async def async_wrapper(*args, **kwargs):
result = await func(*args, **kwargs)
return cls.construct(**kwargs, **result.dict())
return cls.model_construct(**kwargs, **result.dict())
creatorrr marked this conversation as resolved.
Show resolved Hide resolved

# This wrapper handles synchronous functions, directly calling them and processing their results with `cls.construct`.
@wraps(func)
def sync_wrapper(*args, **kwargs):
# Logging at this point might be useful for debugging, but should use a proper logging framework instead of print statements for production code.
result = func(*args, **kwargs)
return cls.construct(**kwargs, **result.dict())
return cls.model_construct(**kwargs, **result.dict())

return async_wrapper if iscoroutinefunction(func) else sync_wrapper

Expand Down
Loading