Created
September 24, 2023 08:14
-
-
Save i701/d35ced6c0f7d1d495c7dc385d87179fb to your computer and use it in GitHub Desktop.
FastAPI tortoise ORM example with created_at and updated_at workaround
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
# pylint: disable=E0611,E0401 | |
from typing import List | |
import pprint | |
from fastapi import FastAPI | |
from models import User_Pydantic, UserIn_Pydantic, Users | |
from pydantic import BaseModel | |
from starlette.exceptions import HTTPException | |
from datetime import datetime | |
from tortoise.contrib.fastapi import register_tortoise | |
app = FastAPI(title="Tortoise ORM FastAPI example") | |
class Status(BaseModel): | |
message: str | |
@app.get("/users", response_model=List[User_Pydantic]) | |
async def get_users(): | |
return await User_Pydantic.from_queryset(Users.all()) | |
@app.post("/users", response_model=User_Pydantic) | |
async def create_user(user: UserIn_Pydantic): | |
user_obj = await Users.create(**user.dict(exclude_unset=True)) | |
return await User_Pydantic.from_tortoise_orm(user_obj) | |
@app.get("/user/{user_id}", response_model=User_Pydantic) | |
async def get_user(user_id: int): | |
return await User_Pydantic.from_queryset_single(Users.get(id=user_id)) | |
@app.put("/user/{user_id}", response_model=User_Pydantic) | |
async def update_user(user_id: int, user: UserIn_Pydantic): | |
existing_data = await User_Pydantic.from_queryset_single(Users.get(id=user_id)) | |
previous_data = existing_data.model_dump() | |
user_data = user.dict(exclude_unset=True) | |
# Here we override the timestamps | |
user_data["modified_at"] = datetime.utcnow() | |
user_data["created_at"] = previous_data["created_at"] | |
pprint.pprint(user_data) | |
await Users.filter(id=user_id).update(**user_data) | |
return await User_Pydantic.from_queryset_single(Users.get(id=user_id)) | |
@app.delete("/user/{user_id}", response_model=Status) | |
async def delete_user(user_id: int): | |
deleted_count = await Users.filter(id=user_id).delete() | |
if not deleted_count: | |
raise HTTPException(status_code=404, detail=f"User {user_id} not found") | |
return Status(message=f"Deleted user {user_id}") | |
register_tortoise( | |
app, | |
db_url="sqlite://db.sqlite3", | |
modules={"models": ["models"]}, | |
generate_schemas=True, | |
add_exception_handlers=True, | |
) |
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
from tortoise import fields, models | |
from tortoise.contrib.pydantic import pydantic_model_creator | |
class Users(models.Model): | |
""" | |
The User model | |
""" | |
id = fields.IntField(pk=True) | |
#: This is a username | |
username = fields.CharField(max_length=20, unique=False) | |
name = fields.CharField(max_length=50, null=True) | |
family_name = fields.CharField(max_length=50, null=True) | |
category = fields.CharField(max_length=30, default="misc") | |
password_hash = fields.CharField(max_length=128, null=True) | |
created_at = fields.DatetimeField(auto_now_add=True) | |
modified_at = fields.DatetimeField(auto_now=True, null=True) | |
def full_name(self) -> str: | |
""" | |
Returns the best name | |
""" | |
if self.name or self.family_name: | |
return f"{self.name or ''} {self.family_name or ''}".strip() | |
return self.username | |
class PydanticMeta: | |
computed = ["full_name"] | |
exclude = ["password_hash"] | |
User_Pydantic = pydantic_model_creator(Users, name="User") | |
# Make sure to exclude `created_at` and `modified_at` to make pydantic happy as of now. Maybe it will be fixed later. | |
# https://docs.pydantic.dev/2.3/errors/validation_errors/#missing | |
UserIn_Pydantic = pydantic_model_creator( | |
Users, name="UserIn", exclude=["created_at", "modified_at"], exclude_readonly=True | |
) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment