0
Help us understand the problem. What are the problem?

posted at

updated at

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 でのテスト

image.png

Register as a new user and use Qiita more conveniently

  1. You can follow users and tags
  2. you can stock useful information
  3. You can make editorial suggestions for articles
What you can do with signing up
0
Help us understand the problem. What are the problem?