- Async SQLAlchemy session
- Custom user class
- Dependencies for specific permissions
- Celery
- Dockerize(Hot reload)
- Event dispatcher
- Cache
> docker-compose -f docker/docker-compose.yml up
> poetry shell
> poetry install
> alembic upgrade head
> python3 main.py --env local|dev|prod --debug
> make test
> make cov
> pre-commit
from core.db import Transactional, session
@Transactional()
async def create_user(self):
session.add(User(email="[email protected]"))
Do not use explicit commit()
. Transactional
class automatically do.
When executing queries concurrently through asyncio.gather()
, you must use the session_factory
context manager rather than the globally used session.
from core.db import session_factory
async def get_by_id(self, *, user_id) -> User:
stmt = select(User)
async with session_factory() as read_session:
return await read_session.execute(query).scalars().first()
async def main() -> None:
user_1, user_2 = await asyncio.gather(
get_by_id(user_id=1),
get_by_id(user_id=2),
)
If you do not use a database connection like session.add()
, it is recommended to use a globally provided session.
Go to core/config.py
and edit WRITER_DB_URL
and READER_DB_URL
in the config class.
If you need additional logic to use the database, refer to the get_bind()
method of RoutingClass
.
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 = Field(None, description="ID")
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
, IsAuthenticated
, AllowAll
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 = Query(10, description="Limit"),
prev: int = Query(None, description="Prev ID"),
):
pass
Insert permission through dependencies
argument.
If you want to make your own permission, inherit BasePermission
and implement has_permission()
function.
Note. In order to use swagger's authorize function, you must put PermissionDependency
as an argument of dependencies
.
Refer the README of https://github.com/teamhide/fastapi-event
from core.helpers.cache import Cache
@Cache.cached(prefix="get_user", ttl=60)
async def get_user():
...
from core.helpers.cache import Cache, CacheTag
@Cache.cached(tag=CacheTag.GET_USER_LIST, ttl=60)
async def get_user():
...
Use the Cache
decorator to cache the return value of a function.
Depending on the argument of the function, caching is stored with a different value through internal processing.
from core.helpers.cache.base import BaseKeyMaker
class CustomKeyMaker(BaseKeyMaker):
async def make(self, function: Callable, prefix: str) -> str:
...
If you want to create a custom key, inherit the BaseKeyMaker class and implement the make() method.
from core.helpers.cache.base import BaseBackend
class RedisBackend(BaseBackend):
async def get(self, key: str) -> Any:
...
async def set(self, response: Any, key: str, ttl: int = 60) -> None:
...
async def delete_startswith(self, value: str) -> None:
...
If you want to create a custom key, inherit the BaseBackend class and implement the get()
, set()
, delete_startswith()
method.
Pass your custom backend or keymaker as an argument to init. (/app/server.py
)
def init_cache() -> None:
Cache.init(backend=RedisBackend(), key_maker=CustomKeyMaker())
from core.helpers.cache import Cache, CacheTag
await Cache.remove_by_prefix(prefix="get_user_list")
await Cache.remove_by_tag(tag=CacheTag.GET_USER_LIST)