Zenn
🧢

FastAPIでログインシステムを作ってみた

2023/09/11に公開
PostgreSQL
FastAPI
Postman
tech

材料

  • FastAPI
  • PostgreSQL

データベースの用意

最初はデータベースとダミーユーザーを作る、ユーザーの追加機能は後回します。

CREATE TABLE users (
	id varchar(128) NOT NULL PRIMARY KEY,  -- 必須、メインキー
	password varchar(256) NOT NULL,            -- 必須
	name varchar(256) DEFAULT 'nanashi'
);

INSERT INTO users VALUES (
	'admin',
	'8c6976e5b5410415bde908bd4dee15dfb167a9c873fc4bb8a81f6f2ab448a918', -- SHA256
	'administrator'
);

FastAPI/RequestBody

FastAPIでは、BaseModelを継承したクラスを経由してRequestBodyを入力。

from pydantic import BaseModel

class LoginUser(BaseModel):
    id: str
    password: str

@app.get("/login/")
async def login(user: LoginUser):
    return {'status': 'OK', 'code': '200', 'message': user}

Postmanを用いてAPIをテスト：

これで下地ができていた。

FastAPI/PostgreSQL

SQLAlchemyを使います。

pip install sqlalchemy

database.py

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "postgresql://%%username%%:%%password%%@%%server%%/%%database%%"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    connect_args = {
        "check_same_thread": False
    }
)
SessionLocal = sessionmaker(
    autocommit = False,
    autoflush = False,
    bind = engine
)

Base = declarative_base()

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, primary_key=True)
    password = Column(String)
    name = Column(String)

schemas.py

from pydantic import BaseModel

class UserBase(BaseModel):
    id: str

class UserLogin(UserBase):
    password: str

class User(UserBase):
    name: str
    class Config:
        from_attributes = True

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_login_user(db: Session, user_id: str):
    return db.query(models.User).filter(models.User.id == user_id).first()

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()

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

@app.post("/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

@app.post("/login/", response_model=schemas.UserLogin)
def login(user: schemas.UserLogin, db: Session = Depends(get_db)):
    login_user = crud.get_login_user(db, user_id=user.id)
    if login_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return login_user

Discussion