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

posted at

updated at

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

こちらのプログラムを改造しました。
FastAPI: SQLAlchemy で MariaDB を使う
変更点
 id を string に
 age を追加

プログラム

フォルダー構造は同じです。

$ 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: str):
    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(id=user.id,email=user.email, hashed_password=fake_hashed_password, name=user.name, age=user.age)
    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: str, 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(String(15), 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)
	age = Column(Integer)
schemas.py
from pydantic import BaseModel

class UserBase(BaseModel):
	email: str

class UserCreate(UserBase):
	id: str
	password: str
	name: str
	age: int

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

	class Config:
		orm_mode = True

テストスクリプト

データの作成

http_create.sh
http http://127.0.0.1:8000/users/ id="t001" email="aa@bbb.com" password="secret" name="大谷" age=25
http http://127.0.0.1:8000/users/ id="t002" email="bb@bbb.com" password="secret" name="菊池" age=28
http http://127.0.0.1:8000/users/ id="t003" email="cc@bbb.com" password="secret" name="鈴木" age=26

データの取得

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/t001

MariaDB の状態

MariaDB [db_test]> show columns from users;
+-----------------+--------------+------+-----+---------+-------+
| Field           | Type         | Null | Key | Default | Extra |
+-----------------+--------------+------+-----+---------+-------+
| id              | varchar(15)  | NO   | PRI | NULL    |       |
| email           | varchar(255) | YES  | UNI | NULL    |       |
| hashed_password | varchar(255) | YES  |     | NULL    |       |
| is_active       | tinyint(1)   | YES  |     | NULL    |       |
| name            | varchar(255) | YES  | MUL | NULL    |       |
| age             | int(11)      | YES  |     | NULL    |       |
+-----------------+--------------+------+-----+---------+-------+
6 rows in set (0.002 sec)

MariaDB [db_test]> select * from users;
+------+------------+-----------------------+-----------+--------+------+
| id   | email      | hashed_password       | is_active | name   | age  |
+------+------------+-----------------------+-----------+--------+------+
| t001 | aa@bbb.com | secretnotreallyhashed |         1 | 大谷   |   25 |
| t002 | bb@bbb.com | secretnotreallyhashed |         1 | 菊池   |   28 |
| t003 | cc@bbb.com | secretnotreallyhashed |         1 | 鈴木   |   26 |
+------+------------+-----------------------+-----------+--------+------+
3 rows in set (0.000 sec)

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?