mirror of
https://github.com/creyD/creyPY.git
synced 2026-04-14 12:20:31 +02:00
Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
40176aa3e9 | ||
|
|
be66bbebbf |
@@ -1,3 +1,2 @@
|
|||||||
from .base import * # noqa
|
from .base import * # noqa
|
||||||
from .response_schema import * #noqa
|
|
||||||
from .schema_optional import * #noqa
|
from .schema_optional import * #noqa
|
||||||
|
|||||||
@@ -11,6 +11,6 @@ class BaseSchemaModelIN(BaseModel):
|
|||||||
|
|
||||||
|
|
||||||
class BaseSchemaModelOUT(BaseSchemaModelIN):
|
class BaseSchemaModelOUT(BaseSchemaModelIN):
|
||||||
id: UUID | str
|
id: UUID
|
||||||
created_at: datetime
|
created_at: datetime
|
||||||
updated_at: datetime
|
updated_at: datetime
|
||||||
|
|||||||
@@ -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
|
|
||||||
@@ -1 +1 @@
|
|||||||
stripe==12.0.0 # Stripe
|
stripe==12.1.0 # Stripe
|
||||||
|
|||||||
Reference in New Issue
Block a user