目次
4日目 - CRUDの作成
前回各モデルの設定が完了したのでそれぞれのエンドポイントで必要な分のCRUDを設定する
api/v1/cruds/user.py
from sqlalchemy.orm import Session
from api.v1.models.user import User
from api.v1.schemas.user as user_schema
def get_user_by_login_id(db:Session,login_id:str):
return db.query(User).filter_by(login_id=login_id).first()
def create_user(db:Session,user:user_schema.UserCreateRequest):
new_user = User(
login_id = user.login_id,
name = user.name,
description = user.description,
password_hash = user.password # あとでハッシュ化させる
)
db.add(new_user)
db.commit()
db.refresh(new_user)
return new_user
api/v1/cruds/post.py
from sqlalchemy.orm import Session
from api.v1.models.post import Post
from api.v1.models.lgtm import Lgtm
from api.v1.schemas.post as post_schema
def get_posts(db:Session):
posts = db.query(Post).all()
for post in posts:
n.lgtm_count = db.query(Lgtm).filter_by(post_id=post.id).count()
return posts
def create_post(db:Session,post:post_schema.PostCreateRequest):
new_post = Post(
title=post.title
context=post.context
user_id=post.user_id
)
db.add(new_post)
db.commit()
db.refresh(new_post)
return new_post
def get_post(db:Session,id:int):
post = db.query(Post).filter_by(id=id).first()
post.lgtm_count = db.query(Lgtm).filter_by(post_id=id).count()
return post
api/v1/cruds/lgtm.py
from sqlalschemy.orm import Session
from api.v1.models.lgtm import Lgtm
def get_lgtm(db:Session,post_id:int,user_id:int):
db_lgtm = db.query(Lgtm).filter(post_id==post_id,user_id==user_id).first()
return db_lgtm
def add_lgtm(db:Session,post_id,user_id:int):
new_lgtm = Lgtm(
post_id=post_id,
user_id=user_id
)
db.add(new_lgtm)
db.commit()
def remove_lgtm(db:Session,lgtm:Lgtm):
db.delete(lgtm)
db.commit()
Routerの設定
CRUDが設定できたのでこれらのメソッドを利用してエンドポイントの処理を入れていく
api/v1/routers/user.py
from fastapi import APIRouter,Depends, HTTPException
from sqlalchemy.orm import Session
import api.v1.schemas.user as user_schema
import api.v1.cruds.user as user_crud
from api.v1.db import get_db
router = APIRouter()
@router.post('/users',response_model=user_schema.User)
def create_user(user:user_schema.UserCreateRequest,db:Session = Depends(get_db)):
db_user = user_crud.get_user_by_login_id(db,user.login_id)
if db_user:
raise HTTPException(status_code=400,detail="LoginId already exists")
return db_user
@router.get('/user/{login_id}',response_model=user_schema.User)
def get_user(login_id:str,db:Session = Depends(get_db)):
db_user = user_crud.get_user_login_id(db,login_id)
if db_user is None:
raise HTTPException(status_code=400,detail = "User not Found")
return db_user
api/v1/routers/post.py
from fastapi import APIRouter, Depends, HTTPException
from typing import List
from sqlalchemy.orm import Session
import api.v1.schemas.post as post_schema
import api.v1.cruds.post as post_cruds
import api.v1.cruds.lgtm as lgtm_cruds
from api.v1.db import get_db
@router.get('/posts',response_model=List[post_schema.Post])
def get_posts(db:Session = Depends(get_db)):
posts = post_cruds.get_posts(db=db)
return posts
@router.post('/posts',response_model=post_schema.Post)
def create_post(post:post_schema.PostCreateRequest,db:Session = Depends(get_db)):
post = post_crud.create_post(db=db,post=post)
return post
@router.get('/post/{id}',response_model=post_schema.Post)
def get_post(id:int,db:Session=Depends(get_db)):
post = post_crud.get_post(db=db,id=id)
if post is None:
raise HTTPException(status_code=400,detail="Post not Found"
return post
@router.post('/post/{id}/lgtm')
def add_lgtm(id:int,db:Session=Depends(get_db)):
lgtm_crud.add_lgtm(db=db,post_id=id,user_id=1) #便宜上user_id=1を指定
return None
@router.delete('/post/{id}/lgtm')
def delete_lgtm(id:int,db:Session=Depends(get_db)):
lgtm = lgtm_crud.get_lgtm(db=db,post_id=id,user_id=1) #便宜上user_id=1を指定
if lgtm is None:
raise HTTPException(status_code=400,detail="Lgtm not Found")
lgtm_crud.remove_lgtm(db=db,lgtm=lgtm)
return None
docsの確認
これでdocsにアクセスして各エンドポイントのリクエスト結果を確認することができるようになる
簡単なAPIとしてはこれでいったん使える