-
Notifications
You must be signed in to change notification settings - Fork 5
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
credit to https://github.com/sfrvn
- Loading branch information
Showing
15 changed files
with
1,212 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -3,3 +3,4 @@ | |
dist/ | ||
fastapi_sqlalchemy_toolkit.egg-info/ | ||
.pypirc | ||
site/ |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -4,3 +4,4 @@ A list of project contributors: | |
|
||
Egor Kondrashov <[email protected]> | ||
Natalia Kamysheva <[email protected]> | ||
Buned Kholmatov <https://github.com/sfrvn> |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,86 @@ | ||
The `ModelManager` methods can be easily extended with additional logic. | ||
|
||
|
||
Firstly, you need to define your own `ModelManager` class: | ||
|
||
```python | ||
from fastapi_sqlalchemy_toolkit import ModelManager | ||
|
||
|
||
class MyModelManager[MyModel, MyModelCreateSchema, MyModelUpdateSchema](ModelManager): | ||
... | ||
``` | ||
|
||
### Additional Validation | ||
You can add additional validation by overriding the `validate` method: | ||
|
||
```python | ||
class MyModelManager[MyModel, MyModelCreateSchema, MyModelUpdateSchema](ModelManager): | ||
async def validate_parent_type(self, session: AsyncSession, validated_data: ModelDict) -> None: | ||
""" | ||
Checks the type of the selected Parent object | ||
""" | ||
# The Parent object with this ID definitely exists since this is checked earlier in super().validate | ||
parent = await parent_manager.get(session, id=in_obj["parent_id"]) | ||
if parent.type != ParentTypes.CanHaveChildren: | ||
raise HTTPException( | ||
status_code=status.HTTP_400_BAD_REQUEST, | ||
detail="This parent has incompatible type", | ||
) | ||
|
||
async def run_db_validation( | ||
self, | ||
session: AsyncSession, | ||
db_obj: MyModel | None = None, | ||
in_obj: ModelDict | None = None, | ||
) -> ModelDict: | ||
validated_data = await super().validate(session, db_obj, in_obj) | ||
await self.validate_parent_type(session, validated_data) | ||
return validated_data | ||
``` | ||
|
||
### Additional business logic for CRUD operations | ||
If additional business logic needs to be executed during CRUD operations with the model, | ||
this can be done by overriding the corresponding `ModelManager` methods: | ||
|
||
```python | ||
class MyModelManager[MyModel, MyModelCreateSchema, MyModelUpdateSchema](ModelManager): | ||
async def create( | ||
self, *args, background_tasks: BackgroundTasks | None = None, **kwargs | ||
) -> MyModel: | ||
created = await super().create(*args, **kwargs) | ||
background_tasks.add_task(send_email, created.id) | ||
return created | ||
``` | ||
|
||
This approach aligns with the "*Fat Models, Skinny Views*" principle from Django. | ||
|
||
### Using declarative dilters in non-standard list queries | ||
If you need to retrieve not just a list of objects but also other fields (e.g., the number of child objects) | ||
or aggregations, and you also need declarative filtering, you can create a new manager method, | ||
calling the `super().get_filter_expression` method within it: | ||
```python | ||
class MyModelManager[MyModel, MyModelCreateSchema, MyModelUpdateSchema](MyModel): | ||
async def get_parents_with_children_count( | ||
self, session: AsyncSession, **kwargs | ||
) -> list[RetrieveParentWithChildrenCountSchema]: | ||
children_count_query = ( | ||
select(func.count(Child.id)) | ||
.filter(Child.parent_id == Parent.id) | ||
.scalar_subquery() | ||
) | ||
query = ( | ||
select(Parent, children_count_query.label("children_count")) | ||
) | ||
|
||
# Calling the method to get SQLAlchemy filters from method arguments | ||
# list и paginated_list | ||
query = query.filter(self.get_filter_expression(**kwargs)) | ||
|
||
result = await session.execute(query) | ||
result = result.unique().all() | ||
for i, row in enumerate(result): | ||
row.Parent.children_count = row.children_count | ||
result[i] = row.Parent | ||
return result | ||
``` |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,49 @@ | ||
### Saving user of the request | ||
|
||
You can associate the user of the request with the object being created/updated | ||
by passing an additional parameter to the `create` (`update`) method: | ||
```python | ||
@router.post("") | ||
async def create_child( | ||
child_in: CreateUpdateChildSchema, session: Session, user: CurrentUser | ||
) -> CreateUpdateChildSchema: | ||
return await child_manager.create(session=session, in_obj=child_in, author_id=user.id) | ||
``` | ||
|
||
### Creating and updating objects with M2M relationships | ||
If the model has an M2M relationship defined, using `ModelManager` allows you to pass a list of object IDs to this field. | ||
|
||
`fastapi-sqlalchemy-toolkit` validates the existence of these objects and establishes the M2M relationship for them, | ||
without the need to create separate endpoints for working with M2M relationships. | ||
|
||
```python | ||
# Let the Person and House models have an M2M relationship | ||
from pydantic import BaseModel | ||
|
||
|
||
class PersonCreateSchema(BaseModel): | ||
house_ids: list[int] | ||
|
||
... | ||
|
||
in_obj = PersonCreateSchema(house_ids=[1, 2, 3]) | ||
await person_manager.create(session, in_obj) | ||
# Creates a Person object and establishes an M2M relationship with Houses with ids 1, 2, and 3 | ||
``` | ||
|
||
### Filtering by list of values | ||
One way to filter by a list of values is to pass this list as a | ||
query parameter in the URL as a comma-separated string. | ||
`fastapi-sqlalchemy-toolkit` provides a utility for filtering by a list of values passed as a comma-separated string: | ||
```python | ||
from uuid import UUID | ||
from fastapi_sqlalchemy_toolkit.utils import comma_list_query, get_comma_list_values | ||
|
||
@router.get("/children") | ||
async def get_child_objects( | ||
session: Session, | ||
ids: comma_list_query = None, | ||
) -> list[ChildListSchema] | ||
ids = get_comma_list_values(ids, UUID) | ||
return await child_manager.list(session, id=FieldFilter(ids, operator="in_")) | ||
``` |
Oops, something went wrong.