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

Adjusted files for isort & autopep

Browse Source
This commit is contained in:
creyD
2025-03-24 08:26:44 +00:00
committed by github-actions[bot]
parent 8363055984
commit c2e2469027
2 changed files with 26 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
creyPY/fastapi/crud.py
View File

@@ -19,14 +19,19 @@ async def get_object_or_404(
db: AsyncSession, db: AsyncSession,
expunge: bool = False, expunge: bool = False,
lookup_column: str = "id", lookup_column: str = "id",
response_fields: List[str] = [] response_fields: List[str] = [],
) -> T: ) -> T:
pass pass
@overload @overload
def get_object_or_404( def get_object_or_404(
db_class: Type[T], id: UUID | str, db: Session, expunge: bool = False, lookup_column: str = "id", response_fields: List[str] = [] db_class: Type[T],
id: UUID | str,
db: Session,
expunge: bool = False,
lookup_column: str = "id",
response_fields: List[str] = [],
) -> T: ) -> T:
pass pass
@@ -37,12 +42,14 @@ def get_object_or_404(
db: Session | AsyncSession, db: Session | AsyncSession,
expunge: bool = False, expunge: bool = False,
lookup_column: str = "id", lookup_column: str = "id",
response_fields: List[str] = [] response_fields: List[str] = [],
) -> T: ) -> T:
async def _get_async_object() -> T: async def _get_async_object() -> T:
if response_fields: if response_fields:
selected_columns = [getattr(db_class, field) for field in response_fields if hasattr(db_class, field)] selected_columns = [
getattr(db_class, field) for field in response_fields if hasattr(db_class, field)
]
query = select(*selected_columns).select_from(db_class) query = select(*selected_columns).select_from(db_class)
else: else:
query = select(db_class).filter(getattr(db_class, lookup_column) == id) query = select(db_class).filter(getattr(db_class, lookup_column) == id)
@@ -56,7 +63,9 @@ def get_object_or_404(
def _get_sync_object() -> T: def _get_sync_object() -> T:
if response_fields: if response_fields:
selected_columns = [getattr(db_class, field) for field in response_fields if hasattr(db_class, field)] selected_columns = [
getattr(db_class, field) for field in response_fields if hasattr(db_class, field)
]
query = db.query(*selected_columns).filter(getattr(db_class, lookup_column) == id) query = db.query(*selected_columns).filter(getattr(db_class, lookup_column) == id)
else: else:
query = db.query(db_class).filter(getattr(db_class, lookup_column) == id) query = db.query(db_class).filter(getattr(db_class, lookup_column) == id)

9
creyPY/fastapi/schemas/response_schema.py
View File

@@ -1,6 +1,8 @@
from typing import List, Optional, Type from typing import List, Optional, Type
from pydantic import BaseModel, create_model from pydantic import BaseModel, create_model
from fastapi import Query from fastapi import Query
class ResponseModelDependency: class ResponseModelDependency:
def __init__(self, model_class: Type[BaseModel]): def __init__(self, model_class: Type[BaseModel]):
self.model_class = model_class self.model_class = model_class
@@ -10,7 +12,7 @@ class ResponseModelDependency:
if not fields: if not fields:
return result return result
if hasattr(result, '_fields'): if hasattr(result, "_fields"):
row_fields = result._fields row_fields = result._fields
return dict(zip(row_fields, result)) return dict(zip(row_fields, result))
elif isinstance(result, tuple): elif isinstance(result, tuple):
@@ -25,7 +27,7 @@ class ResponseModelDependency:
all_annotations = {} all_annotations = {}
for cls in self.model_class.__mro__: for cls in self.model_class.__mro__:
if hasattr(cls, '__annotations__'): if hasattr(cls, "__annotations__"):
all_annotations.update(cls.__annotations__) all_annotations.update(cls.__annotations__)
fields = {} fields = {}
@@ -33,7 +35,6 @@ class ResponseModelDependency:
if field in all_annotations: if field in all_annotations:
fields[field] = (all_annotations[field], None) fields[field] = (all_annotations[field], None)
dynamic_model = create_model(f'Dynamic{self.model_class.__name__}', **fields) dynamic_model = create_model(f"Dynamic{self.model_class.__name__}", **fields)
return dynamic_model, response_fields, process_result return dynamic_model, response_fields, process_result