FastAPI: SQLAlchemy で MariaDB を使う (その 4)

こちらのプログラムを改造しました。
FastAPI: SQLAlchemy で MariaDB を使う (その 3)

API に PUT を追加

プログラム

フォルダー構造

$ tree
.
├── __init__.py
├── crud.py        変更
├── database.py
├── main.py        変更
├── models.py
└── schemas.py     変更

crud.py

from sqlalchemy.orm import Session
import sys
from datetime import datetime

from . import models, schemas
#
def get_city(db: Session, city_id: str):
	return db.query(models.City).filter(models.City.id == city_id).first()
#
def get_cities(db: Session, skip: int = 0, limit: int = 100):
	return db.query(models.City).offset(skip).limit(limit).all()
#
def create_city(db: Session, city: schemas.CityCreate):
	db_city = models.City(id=city.id, name=city.name, \
		population=city.population, date_mod=city.date_mod)
	db.add(db_city)
	db.commit()
	db.refresh(db_city)
	return db_city
#
def update_city(db: Session, city_up: schemas.CityUpdate):
	city_id = city_up.id
	db_city = db.query(models.City).filter(models.City.id == city_id).first()
	db_city.population = city_up.population
	db_city.date_mod = datetime.now()
	db.commit()
	db.refresh(db_city)
	return db_city
#
def delete_city(db: Session, city_id: str):
	db_city = db.query(models.City).filter(models.City.id == city_id).first()
	db.delete(db_city)
	db.commit()
#

main.py

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine
import sys

models.Base.metadata.create_all(bind=engine)

app = FastAPI()

# Dependency
def get_db():
	db = SessionLocal()
	try:
		yield db
	finally:
		db.close()
#
@app.post("/cities/", response_model=schemas.City)
def create_city(city: schemas.CityCreate, db: Session = Depends(get_db)):
	return crud.create_city(db=db, city=city)
#
@app.put("/cities/", response_model=schemas.City)
def update_city(city_up: schemas.CityUpdate, db: Session = Depends(get_db)):
	crud.update_city(db=db, city_up=city_up)
	db_city = crud.get_city(db, city_id=city_up.id)
	return db_city
#
@app.get("/cities/", response_model=list[schemas.City])
def read_cities(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
	cities = crud.get_cities(db, skip=skip, limit=limit)
	return cities
#
@app.get("/cities/{city_id}", response_model=schemas.City)
def read_city(city_id: str, db: Session = Depends(get_db)):
	db_city = crud.get_city(db, city_id=city_id)
	if db_city is None:
		raise HTTPException(status_code=404, detail="City not found")
	return db_city
#
@app.delete("/cities/{city_id}")
def delete_city(city_id: str, db: Session = Depends(get_db)):
	crud.delete_city(db, city_id=city_id)
#

schemas.py

#
from datetime import datetime
from pydantic import BaseModel

class CityBase(BaseModel):
	id: str

class CityCreate(CityBase):
	id: str
	name: str
	population: int
	date_mod: datetime

class CityUpdate(CityBase):
	id: str
	population: int

class City(CityBase):
	id: str
	name: str
	population: int
	date_mod: datetime

	class Config:
		orm_mode = True

テストスクリプト

データの更新

http_update.sh

http PUT http://127.0.0.1:8000/cities/ id=t3324 population=2598300

GUI でのテスト