Skip to content

Instantly share code, notes, and snippets.

@i701
Created September 24, 2023 08:14
Show Gist options
  • Save i701/d35ced6c0f7d1d495c7dc385d87179fb to your computer and use it in GitHub Desktop.
Save i701/d35ced6c0f7d1d495c7dc385d87179fb to your computer and use it in GitHub Desktop.
FastAPI tortoise ORM example with created_at and updated_at workaround
# 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,
)
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