creyD/creyPY
1
0
Fork 0
mirror of https://github.com/creyD/creyPY.git synced 2026-04-12 19:30:30 +02:00
Code Issues Packages Projects Releases Wiki Activity

fix: remove optional schema and alter base out (#49)

Browse Source 
* fix: remove optional schema and alter base out

* changes
This commit is contained in:
vikynoah
2025-04-15 20:32:02 +02:00
committed by GitHub
parent 79dde8008a
commit be66bbebbf
3 changed files with 1 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
creyPY/fastapi/schemas/__init__.py
View File

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

2
creyPY/fastapi/schemas/base.py
View File

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

43
creyPY/fastapi/schemas/response_schema.py
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