Fastapi mc style tools to make it easier to develop.
Project description
FastAPI-MCtools
- MC(나)의 FastAPI로 개발할 때 자주 사용하는 코드들을 모아놓은 패키지입니다.
- This is a package that contains frequently used codes when developing with FastAPI by MC(me).
- This is kind of an utility package
- 블로그 소개글 참고
Installation
pip install fastapi-mctools
Usage
CLI
1-1. run dev
- run uvicorn server for development, this command wraps uvicorn command.
- this command will find
main.py
and run uvicorn server - use this command when running in local environment and use --port and --host for setting port and host. default is 8000 and 127.0.0.1
mct run dev
1-2. run prod
- run gunicorn server for production, this command wraps gunicorn command.
- this command will find
gunicorn.config.py
and run gunicorn server - if not existing
gunicorn.config.py
, it will make basic gunicorn config file
mct run prod
2. startproject
- create a new FastAPI project in ordinary way
mct startproject
3. gunicorn
- make gunicorn basic config
mct gunicorn
4. types
- check all functions if existing missing type hint
- this command will find all functions in the file or directory and check if there is any missing type hint
- use this if you are very strict about type hinting
mct types <path: directory or file>
5. shell
- simple ipython shell
- need to install ipython
mct shell
# in ipython shell
# need to import session that made by generator or async generator
# sync session
from app.session import get_db
db = next(get_db())
# async session
from app.session import get_db
db = await anxet(get_db())
DB Session
- DB, AsyncDB
- when you want to make db session in the fast way, use this
from fastapi_mctools.db import DB, AsyncDB
get_db = DB(db_url)
get_db = AsyncDB(db_url)
Request Logging Middleware
- RequestLoggingMiddleware
- when you simply want to log all requests, use this
- you can use log configuration by dictConfig or fileConfig
# main.py
import logging
from fastapi import FastAPI
from fastapi_mctools.middlewares import RequestLoggingMiddleware
app = FastAPI()
logger = logging.getLogger('request')
app.add_middleware(RequestLoggingMiddleware, logger=logger)
SQLAlchemy ORM
- sync_base, async_base
- these are kind of repositories for frequently used ORM operations, such as CRUD
- if not putting any column name, it will use all columns in the model
- can use filterbackend for filtering data
from fastapi_mctools.orms.sqlalchemy import sync_base, async_base
from fastapi_mctools.orms.filters import FilterBackend
class UserCreate(async_base.ACreateBase):
...
class UserRead(async_base.AReadBase):
...
class UserUpdate(async_base.AUpdateBase):
...
class UserDelete(async_base.ADeleteBase):
...
class UserRepository(UserCreate, UserRead, UserUpdate, UserDelete):
def __init__(self, model: SqlalchemyModel) -> None:
super().__init__(model=model)
user_repository = UserRepository(model=User)
# AsyncSession from sqlalchemy.ext.asyncio
async def create_user(db: AsyncSession, data: dict) -> User:
return await user_repository.create(db, **data)
async def read_user(db: AsyncSession, user_id: int) -> User:
user = await user_repository.get_by_id(db, user_id)
users = await user_repository.get_by_filters(db, age=20) # This will return data for users whose age is 20.
...
async def read_user_with_filterbackend(db: AsyncSession, user_id: int) -> User:
filter_backend = FilterBackend()
filter_backend.add_filter("age", 20)
filter_backend.add_filter("name", "%test%")
users = await user_repository.get_by_filters(db, filter_backend=filter_backend) # This will return data for users whose age is 20 and name contains "test"
...
# update and delete are similar to create and read
TestTools
- db_managers
- when you want to test with db and set up and tear down db in very simple way
# conftest.py
from fastapi_mctools.test_tools.db_managers import TestConfDBManager
db_test_manager = TestConfDBManager(TEST_DB_URL)
@pytest.fixture
async def db():
# Base is your sqlalchemy base
async for session in db_test_manager.get_async_db_session(base=Base, is_meta=True):
yield session
# test_something.py
async def test_something(db):
# db is your test db session
...
dependencies
- Dependency
- this is a class that can be used as a dependency manager that can handle multiple dependencies in an single instance
- so you can only import one instance even if there are many dependencies
async def temp_dep_1():
return 1
async def temp_dep_2():
return 2
async def temp_dep_3():
return 3
temp_dep = Dependency(
TempDep1=temp_dep_1,
TempDep2=temp_dep_2,
TempDep3=temp_dep_3
)
# in your route
@app.get('/temp_1')
async def temp_1(result: temp_dep.TempDep1):
return result
@app.get('/temp_2')
async def temp_2(result: temp_dep.TempDep2):
return result
@app.get('/temp_3')
async def temp_3(result: temp_dep.TempDep3):
return result
- create_simple_form_dependency
- this is developed to avoid the tedious task of repeatedly writing out the Form.
- especially when you make api with file upload, you can use this to make it simple
from fastapi_mctools.dependencies import create_simple_form_dependency
simple_form_params ={
"name": str,
"age": int,
"address": str,
"email": str,
"status": str,
"memo": str,
}
form_data = create_simple_form_dependency(simple_form_params)
@app.post("/user")
def create_user(form_data: FastAPIForm = Depends(), file: UploadFile = File()):
return form_data
Requests
- APIClient
- this is a class that can be used to make requests to external APIs only asynchronously
- you can keep the session alive and reuse it
- I normally use this in the lifespan of the app
from fastapi_mctools.utils.requests import APIClient
@asynccontextmanager
async def lifespan(app: FastAPI):
try:
api_client = APIClient()
await api_client.start()
states["api_client"] = api_client
logger.info(f"Start App: states {states}, Debug: {app_settings.DEBUG}")
yield
await api_client.close()
except Exception as e:
raise e
Responses
- ResponseInterFace
- this is an interface that can unify the response format ex) {"results" : data, "message": message, "status": status}
- it's not compatible with
response_model
in router.
from fastapi_mctools.utils.responses import ResponseInterFace
async def get_user(user_id: int) -> ResponseInterFace:
user = await user_repository.get(user_id)
if user:
return ResponseInterFace(result=user, message="Success", temp_response_1="temp1", temp_response_2="temp2", status=200)
return ResponseInterFace(result=None, message="Not Found", temp_response_1="temp1", temp_response_2="temp2", status=404)
from pydantic import BaseModel
class UserResponse(BaseModel)
# Let's assume that the model is defined
...
@router.get("/user/{user_id}")
async def get_user(user_id: int):
...
response = ResponseInterFace(result=user, message="Success", temp_response_1="temp1", temp_response_2="temp2", status=200)
return UserResponse(**response)
time
- time_checker
- this is a decorator that can be used to check orm query time
from fastapi_mctools.utils.time import time_checker
@time_checker(debug=True, logger=logger)
class SomeRepository:
def get(self, db: AsyncSession, id: int) -> SomeModel:
...
temp = SomeRepository()
temp.get(db, 1) # SomeRepository.get took 0.01 ms
exceptions
- HTTPException, handle_http_exception, handle_500_exception
- HTTPException is a class that can be used instead of FastAPI's HTTPException, you can add more information to the exception response
- handle_http_exception is exception handler that can be used to handle HTTPException, if you want to add more information with HTTPException, add this handler to app
- handle_500_exception is exception handler that can be used to handle 500 error, it converts Exception to HTTPException and returns it
from fastapi_mctools.exceptions import HTTPException, handle_http_exception, handle_500_exception
async def temp_api():
raise HTTPException(status_code=400, detail="Bad Request", more_info="more info", code="TEMP_ERROR")
app.add_exception_handler(HTTPException, handle_http_exception)
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
fastapi_mctools-0.4.0.tar.gz
(24.1 kB
view hashes)
Built Distribution
Close
Hashes for fastapi_mctools-0.4.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 3282001b651458e355fc3fa7a919bec24fbc57fe08943f48dc156cd2b3c83718 |
|
MD5 | 0f561c92e8aba61fb2e8440ad7ee6313 |
|
BLAKE2b-256 | a97ee3ed02c97d001fd45a9da77a872cf2203ca4f7a20abdbfc97498109660b3 |