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

posted at

updated at

FastAPI: SQLAlchemy で MariaDB を使う

次のページを参考にしました。
SQL (Relational) Databases

Ubuntu でのライブラリーのインストール

sudo apt install libmysqlclient-dev
sudo pip3 install mysqlclient

MariaDB に次のデータベースを作成します。

host ='localhost'
data_base = 'db_test'
user ='scott'
password = 'tiger123'

プログラム

$ tree
.
├── __init__.py
├── crud.py
├── database.py
├── main.py
├── models.py
└── schemas.py
crud.py
from sqlalchemy.orm import Session

from . import models, schemas

def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()

def get_user_by_email(db: Session, email: str):
    return db.query(models.User).filter(models.User.email == email).first()

def get_users(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.User).offset(skip).limit(limit).all()

def create_user(db: Session, user: schemas.UserCreate):
    fake_hashed_password = user.password + "notreallyhashed"
    db_user = models.User(email=user.email, hashed_password=fake_hashed_password, name=user.name)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user
database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
#
host ='localhost'
data_base = 'db_test'
user ='scott'
password = 'tiger123'
#
db_url = "mysql://" + user + ":" + password + "@" + host + "/" + data_base
#

engine = create_engine(db_url)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()
main.py
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

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

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

app = FastAPI()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=list[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user
models.py
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship

from .database import Base


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String(255), unique=True, index=True)
    hashed_password = Column(String(255))
    is_active = Column(Boolean, default=True)
    name = Column(String(255), index=True)
schemas.py
from pydantic import BaseModel

class UserBase(BaseModel):
    email: str

class UserCreate(UserBase):
    password: str
    name: str

class User(UserBase):
    id: int
    is_active: bool
    name: str

    class Config:
        orm_mode = True

サーバーの実行

uvicorn sql_app.main:app --reload

テストスクリプト

データの作成

http_create.sh
http 'http://127.0.0.1:8000/users/' email="aa@bbb.com" password="secret" name="Tom"
http 'http://127.0.0.1:8000/users/' email="bb@bbb.com" password="secret" name="Betty"
http 'http://127.0.0.1:8000/users/' email="cc@bbb.com" password="secret" name="John"

データの取得

http_get.sh
http 'http://127.0.0.1:8000/users/?skip=0&limit=100'

id を指定してデータの取得

http_user.sh
http 'http://127.0.0.1:8000/users/1'

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
1
Help us understand the problem. What are the problem?