- SQlAlchemy session
- Custom user class
- Top-level dependency
- Dependencies for specific permissions
- Celery
- Dockerize(Hot reload)
from core.db import session
Just import session and use it.
To guarantee of transaction, session's autocommit
option is True
.
So you have to use Transaction
class.
from core.db import Transaction, session
@Transaction()
async def create_user(self):
session.add(User(email="[email protected]"))
Usage as decorator.
from core.db import Transaction, session
with Transaction():
session.add(User(email="[email protected]"))
Usage as context manager.
In this case, only one transaction is supported.
Note. Do not use explicit commit()
. Transaction
class automatically do.
According to the current settings, the session is set through middleware.
However, it doesn't go through middleware in tests or background tasks.
So you need to use the @create_session
decorator.
from core.db import create_session
@create_session
def test_something():
...
from fastapi import Request
@home_router.get("/")
def home(request: Request):
return request.user.id
Note. you have to pass jwt token via header like Authorization: Bearer 1234
Custom user class automatically decodes header token and store user information into request.user
If you want to modify custom user class, you have to update below files.
core/fastapi/schemas/current_user.py
core/fastapi/middlewares/authentication.py
class CurrentUser(BaseModel):
id: int = None
Simply add more fields based on your needs.
current_user = CurrentUser()
After line 18, assign values that you added on CurrentUser
.
Note. Available from version 0.62 or higher.
Set a callable function when initialize FastAPI() app through dependencies
argument.
Refer Logging
class inside of core/fastapi/dependencies/logging.py
Permissions IsAdmin
and IsAuthenticated
have already been implemented.
from core.fastapi.dependencies import (
PermissionDependency,
IsAdmin,
)
user_router = APIRouter()
@user_router.get(
"",
response_model=List[GetUserListResponseSchema],
response_model_exclude={"id"},
responses={"400": {"model": ExceptionResponseSchema}},
dependencies=[Depends(PermissionDependency([IsAdmin]))], # HERE
)
async def get_user_list(limit: int = 10, prev: int = None):
pass
Insert permission through dependencies
argument.
If you want to make your own permission, inherit BasePermission
and implement has_permission()
function.