Compare commits

..

4 Commits

Author SHA1 Message Date
vikynoah
be66bbebbf fix: remove optional schema and alter base out (#49)
* fix: remove optional schema and alter base out

* changes
2025-04-15 20:32:02 +02:00
renovate[bot]
79dde8008a feat(deps): update dependency stripe to v12 (#42)
Co-authored-by: renovate[bot] <29139614+renovate[bot]@users.noreply.github.com>
2025-04-09 14:01:08 +02:00
creyD
adb017d6ce Adjusted files for isort & autopep 2025-04-09 12:01:00 +00:00
vikynoah
e160cc5fea feat: Response filter async (#47)
* fix: get_object alter for async response filter

* fix: Alter async response

* feat: Decorator for schema out
2025-04-09 14:00:30 +02:00
6 changed files with 22 additions and 47 deletions

View File

@@ -67,7 +67,6 @@ def get_object_or_404(
row = result.scalar_one_or_none()
if row is None:
raise HTTPException(status_code=404, detail="The object does not exist.")
obj_dict = row
if expunge:
await db.expunge(obj_dict)

View File

@@ -1,2 +1,2 @@
from .base import * # noqa
from .response_schema import * #noqa
from .schema_optional import * #noqa

View File

@@ -11,6 +11,6 @@ class BaseSchemaModelIN(BaseModel):
class BaseSchemaModelOUT(BaseSchemaModelIN):
id: UUID | str
id: UUID
created_at: datetime
updated_at: datetime

View File

@@ -1,43 +0,0 @@
from typing import List, Optional, Type
from fastapi import Query
from pydantic import BaseModel, create_model
class ResponseModelDependency:
def __init__(self, model_class: Type[BaseModel]):
self.model_class = model_class
def __call__(self, response_fields: Optional[List[str]] = Query(None)) -> Type[BaseModel]:
def process_result(result, fields=None, async_session=False):
if not fields:
if async_session:
return {k: v for k, v in result.__dict__.items() if not k.startswith("_")}
return result
if hasattr(result, "_fields"):
row_fields = result._fields
return dict(zip(row_fields, result))
elif isinstance(result, tuple):
return dict(zip(fields, result))
elif isinstance(result, dict):
return result
else:
return {field: getattr(result, field) for field in fields if hasattr(result, field)}
if not response_fields:
return self.model_class, None, process_result
all_annotations = {}
for cls in self.model_class.__mro__:
if hasattr(cls, "__annotations__"):
all_annotations.update(cls.__annotations__)
fields = {}
for field in response_fields:
if field in all_annotations:
fields[field] = (all_annotations[field], None)
dynamic_model = create_model(f"Dynamic{self.model_class.__name__}", **fields)
return dynamic_model, response_fields, process_result

View File

@@ -0,0 +1,19 @@
from typing import Optional, Type, Union, get_args, get_origin, get_type_hints
from pydantic import BaseModel, create_model
def optional_fields(cls: Type[BaseModel]) -> Type[BaseModel]:
fields = {}
for name, hint in get_type_hints(cls).items():
if name.startswith("_"):
continue
if get_origin(hint) is not Union or type(None) not in get_args(hint):
hint = Optional[hint]
fields[name] = (hint, None)
new_model = create_model(cls.__name__, __base__=cls, **fields)
return new_model

View File

@@ -1 +1 @@
stripe==11.6.0 # Stripe
stripe==12.0.0 # Stripe