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

Adjusted files for isort & autopep

Browse Source
This commit is contained in:
creyD
2024-11-15 11:39:59 +00:00
committed by github-actions[bot]
parent 9bba5b0a4e
commit 6f09c2ef4c
1 changed files with 41 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
creyPY/fastapi/crud.py
View File

@@ -11,28 +11,31 @@ from .models.base import Base
T = TypeVar("T", bound=Base)
@overload
async def get_object_or_404(
db_class: Type[T],
id: UUID | str,
db: AsyncSession,
expunge: bool = False,
lookup_column: str = "id"
lookup_column: str = "id",
) -> T:
pass
@overload
def get_object_or_404(
db_class: Type[T],
id: UUID | str,
db: Session,
expunge: bool = False,
lookup_column: str = "id"
db_class: Type[T], id: UUID | str, db: Session, expunge: bool = False, lookup_column: str = "id"
) -> T:
pass
def get_object_or_404(
db_class: Type[T], id: UUID | str, db: Session | AsyncSession, expunge: bool = False, lookup_column: str = "id"
db_class: Type[T],
id: UUID | str,
db: Session | AsyncSession,
expunge: bool = False,
lookup_column: str = "id",
) -> T:
async def _get_async_object() -> T:
@@ -61,7 +64,6 @@ def get_object_or_404(
raise HTTPException(status_code=404, detail="Invalid session type. Expected Session or AsyncSession.") # type: ignore
# TODO: Add testing
@overload
async def create_obj_from_data(
@@ -69,26 +71,20 @@ async def create_obj_from_data(
model: Type[T],
db: AsyncSession,
additional_data: dict = {},
exclude: dict = {}
exclude: dict = {},
) -> T:
pass
@overload
def create_obj_from_data(
data: BaseModel,
model: Type[T],
db: Session,
additional_data: dict = {},
exclude: dict = {}
data: BaseModel, model: Type[T], db: Session, additional_data: dict = {}, exclude: dict = {}
) -> T:
pass
def create_obj_from_data(
data: BaseModel,
model: Type[T],
db: Session | AsyncSession,
additional_data={},
exclude={}
data: BaseModel, model: Type[T], db: Session | AsyncSession, additional_data={}, exclude={}
) -> T:
obj_data = data.model_dump(exclude=exclude) | additional_data
obj = model(**obj_data)
@@ -123,10 +119,11 @@ async def update_obj_from_data(
partial: bool = True,
ignore_fields: list = [],
additional_data: dict = {},
exclude: dict = {}
exclude: dict = {},
) -> T:
pass
@overload
def update_obj_from_data(
data: BaseModel,
@@ -136,10 +133,11 @@ def update_obj_from_data(
partial: bool = True,
ignore_fields: list = [],
additional_data: dict = {},
exclude: dict = {}
exclude: dict = {},
) -> T:
pass
def update_obj_from_data(
data: BaseModel,
model: Type[T],
@@ -179,23 +177,19 @@ def update_obj_from_data(
else:
raise HTTPException(status_code=404, detail="Invalid session type. Expected Session or AsyncSession.") # type: ignore
# TODO: Add testing
@overload
async def delete_object(
db_class: Type[T], id: UUID | str, db: AsyncSession
) -> None:
async def delete_object(db_class: Type[T], id: UUID | str, db: AsyncSession) -> None:
pass
@overload
def delete_object(
db_class: Type[T], id: UUID | str, db: Session
) -> None:
def delete_object(db_class: Type[T], id: UUID | str, db: Session) -> None:
pass
def delete_object(
db_class: Type[T], id: UUID | str, db: Session | AsyncSession
) -> None:
def delete_object(db_class: Type[T], id: UUID | str, db: Session | AsyncSession) -> None:
async def _delete_async_obj() -> None:
query = select(db_class).filter(db_class.id == id)
result = await db.execute(query)