Search

SQL Alchemy

1. Basic

가. Set Up

1) SQLite

•
connect_args 매개변수는 SQLite에만 사용됨
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db" engine = create_engine( SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False} ) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base()
Python
복사

2) PostreSQL

SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db" engine = create_engine(SQLALCHEMY_DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base()
Python
복사

나. 동작 흐름

1) set database

•
ORM 설정

2) define database models

•
database와 맵핑되는 model 정의

3) define pydantic models

•
database와는 별개로 어플리케이션에서 활용하기 위한 model 정의

4) create repository functions

•
HTTP methods와는 별개로 도메인과 DB 간의 상호작용에 대한 함수를 정의하여 도메인와 DB 사이의 연결성을 약하게 한다
→ 연결성이 약해지면 도메인에 대한 단위테스트 작성이 보다 용이해짐
•
db model을 기준으로 DB에서 데이터 조회
# models == db models def get_user(db: Session, user_id: int): return db.query(models.User).filter(models.User.id == user_id).first()
Python
복사
•
pydantic model을 기준으로 웹 어플리케이션에서 데이터를 가공하여 DB에 저장함
→ commit(): DB 갱신
→ refresh(): DB 갱신에 따른 model instance(db_user) 갱신
ex) DB에서 생성되는 auto increasing id 등이 DB 갱신에 따라 db_user에 반영됨
# schemas = pydantic models 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) db.add(db_user) db.commit() db.refresh(db_user) return db_user
Python
복사

5) use in application

•
retrieve certain domain
@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
Python
복사
•
list domains
@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
Python
복사
•
create domain
@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)
Python
복사

2. Practice

가. 기본 조작

1) 목표

•
스크립트 실행에 따라 source data 가공하여 아래와 같은 출력을 얻어보자
### All movies: The Bourne Identity was released on 2002-10-11 Furious 7 was released on 2015-04-02 No Pain No Gain was released on 2013-08-23 ### Recent movies: Furious 7 was released after 2015 ### Dwayne Johnson movies: The Rock starred in No Pain No Gain The Rock starred in Furious 7 ### Actors that live in Glendale: Dwayne Johnson has a house in Glendale Mark Wahlberg has a house in Glendale
Python
복사
•
source data
# movies bourne_identity, "The Bourne Identity", date(2002, 10, 11) furious_7, "Furious 7", date(2015, 4, 2) pain_and_gain, "Pain & Gain", date(2013, 8, 23) # actors matt_damon, "Matt Damon", date(1970, 10, 8) dwayne_johnson, "Dwayne Johnson", date(1972, 5, 2) mark_wahlberg, "Mark Wahlberg", date(1971, 6, 5) # contact details of actors matt_contact, "415 555 2671", "Burbank, CA", matt_damon dwayne_contact, "423 555 5623", "Glendale, CA", dwayne_johnson dwayne_contact_2, "421 444 2323", "West Hollywood, CA", dwayne_johnson mark_contact, "421 333 9428", "Glendale, CA", mark_wahlberg # stuntmen matt_stuntman, "John Doe", True, matt_damon dwayne_stuntman, "John Roe", True, dwayne_johnson mark_stuntman, "Richard Roe", True, mark_wahlberg
Python
복사

2) Define DB Models

3) Define Pydantic Models

4) Create Repository Functions

5) Use in App

Reference

•
FastAPI Official Document