[관리자 페이지] Admin Page - 회원가입 구현하기 #3 Insert User Data

[관리자 페이지] Admin Page - 회원가입 구현하기 #3 Insert User Data

---

현재 user가 id와 password만 api에 제공하면 데이터가 insert 되야 합니다.

이 과정을 진행 하기 위해서는 sqlAlchemy에서 session을 얻어와 FastAPI DI를 이용해 실행되는 api method에 주입시켜 줘야합니다.

먼저 session을 얻어오는 함수를 core.py에 작성해 줍니다.

# database/core.py
from typing import Generator

from sqlalchemy import URL, create_engine
from sqlalchemy.orm import sessionmaker, Session

url = URL.create(drivername="mariadb+mariadbconnector", username="hiio420", password="...", host="...",
                 port="...", database="HIIO_ADMIN")

engine = create_engine(url)

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

def get_db() -> Generator[Session, None, None]:
    with SessionLocal() as session:
        yield session

sessionmaker를 통해 SessionLocal을 정의하고 get_db 함수를 만들어 with 과 yield 를 이용해 session을 이용합니다.

https://docs.python.org/ko/dev/reference/expressions.html#yield-expressions

이제 signup과 signin 함수에 di를 하기 위해 파라미터 값으로 Depends를 이용해 넘겨 줍니다.

# api/api_main.py
from typing import Optional

from fastapi import APIRouter, Depends
from sqlalchemy import insert
from sqlalchemy.orm import Session

from api.models import UserModel
from database import get_db, User

api_main = APIRouter()


@api_main.post("/signin")
def signin(user:UserModel,session:Session = Depends(get_db)):
    return {"accessToken": "accessToken", "refreshToken": "refreshToken"}


@api_main.post("/signup")
def signup(user:UserModel,session:Session = Depends(get_db)):
    return {"message": "success"}

db에 user 정보를 저장하는 service 를 하나 만들어 줍니다.

.
┣━api
┣━database
┣━service
┃  ┣━__init__.py
┃  ┣━user.py

root 디렉토리에 service directory를 만들고 그안이 init 파일과 user 파일을 만들어 줍니다.

UserService class를 만들어 주는데 생성자로 db Genertor 객체를 받는(get_db)서비스를 만들어 주고
method로 insert_user를 정의해 줍니다.

# service/user.py
from typing import Generator

from sqlalchemy import insert
from sqlalchemy.orm import Session

from api.models import UserModel
from database import User


class UserService:

    def __init__(self, db: Generator[Session, None, None]):
        self.i = 0
        self.db = next(db)

    def insert_user(self, user: UserModel) -> bool:
        try:
            self.db.add(User(**user.dict()))
            self.db.commit()
            return True
        except Exception as e:
            self.db.rollback()
            return False

이제 __init__.py 에 UserService를 초기화 하여 userService 변수에 할당하고 all을 이용해서 다른 파일에서도 사용할 수 있게끔 해줍니다.

# service/__init__.py
from database import get_db
from .user import UserService

userService = UserService(get_db())

__all__ = ['userService']

api_main.py 파일의 signup 함수에 userSerivce의 insert_user를 호출합니다.

# api/api_main.py

from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session

from api.models import UserModel
from database import get_db
from service import userService
api_main = APIRouter()


@api_main.post("/signin")
def signin(user: UserModel, session: Session = Depends(get_db)):
    return {"accessToken": "accessToken", "refreshToken": "refreshToken"}


@api_main.post("/signup")
def signup(user: UserModel):
    status = userService.insert_user(user)
    return {"message": f"Insert {status}"}

서버를 실행하고 docs 다시 접속 해봅니다.

 

데이터가 잘 된것을 볼 수 있습니다.