프로젝트 진행하면서 Python DB랑 연결할 라이브러리로 SQLAlchemy을 가져다 쓰고있다.

Pydantic을 결합한 래퍼 라이브러리인 SQLModel도 있다고 하지만,

기왕 프로젝트 하는거 기본적인 라이브러리부터 쓰자 싶어서 가져다 써보고 있다.

SQLAlchemy 계층구조

SQLAlchemy는 두 가지 계층으로 나뉜다.

Core(SQL Expression Language):

SQL이 들어갔듯이 데이터베이스와 가까운 저수준 API다.

Python 코드로 SQL 쿼리 자체를 추상화 해서 작성하는데,

select(users_table).where(...)이런 방식으로 작성을 해서 사용하는거다.

ORM보다 빠르고 구체적인 제어가 필요한 경우 사용한다.

ORM(Object Relational Mapper)

Core를 기반으로 구축된 고수준 API다.

Object, Mapper를 보면 알 수 있듯이 데이터베이스 테이블을 Python 모델로 매핑한다.

즉 Python 모델을 제어하는 것으로 SQL이 아닌 객체를 다루듯이 데이터를 조작하는 것이다.

구축 방법

SQLAlchemy 사용을 위해서는 엔진(Engine)과 세션(Session)을 설정해야한다.

Engine은 데이터베이스와의 실제 연결을 관리하는 공장이고

Session은 ORM에서 데이터베이스와 대화하는 작업 공간을 생각하면 된다.

기본적인 설치는 했다고 생각하고 넘어가고, 코드 작살날까봐 먼저 분리해서 설명하겠다.

DATABASE_URL = "postgresql://user:password@localhost/dbname" engine = create_engine(DATABASE_URL, echo=True)

라이브러리에서 제공되는 create_engine에 DB Url 정보를 넘기면 엔진이 생성된다.

이때 echo를 True로 두면 실행 로그를 출력한다.

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

생성된 엔진을 넘겨서 sessionmaker를 생성한다.

maker라는 이름에서 예상되듯이 팩토리 패턴으로 세션에 대한 생성을 담당한다.

Base = declarative_base()

모델을 사용하기 위해 모델들의 기본이 될 Base를 생성한다.

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

라인 유지용 임시텍스트

의존성 주입용 제너레이터 함수다.

FastAPI에서 권장되는 주입 방식으로

요청 단위로 세션을 주입한다.

사용예시 @app.get("/items/") def read_items(db: Session = Depends(get_db)):

yield db 방식으로 비즈니스 로직이 실행되는 동안 세션을 유지시킨다.

모델 정의

기본적인 구조를 구축 했으면 모델을 정의하는데,

먼저 테이블과 매핑될 모델을 정의하고 테이블 명을 입력한다.

class User(Base): """사용자 기본 정보 및 OAuth 인증"""

tablename = "users" table_args = ( UniqueConstraint('oauth_provider', 'oauth_id', name='uq_users_oauth'), )

라인 유지용 임시텍스트

__tablename__ 을 통해 테이블과 매핑되고

__table_args__에서는 제약조건 등의 별도 테이블 속성이 입력된다.

id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) oauth_provider = Column(String(50), nullable=False) # OAuth 제공자 (google, naver, kakao) oauth_id = Column(String(255), nullable=False) # OAuth 제공자의 사용자 ID email = Column(String(255), unique=True, nullable=False, index=True) # 이메일 주소 username = Column(String(100), nullable=False) # 사용자 이름

라인 유지용 임시 텍스트

실제 컬럼들은 Column 매핑을 통해 처리되는데 각 컬럼별로 기본키, null 여부, 속성, 인덱싱 여부 등 별도의 조건을 매개변수로 설정 가능하다.

관계

profile = relationship("UserProfile", back_populates="user", uselist=False, cascade="all, delete-orphan") conversations = relationship("Conversation", back_populates="user")

라인 유지용 임시 텍스트

여러 테이블간의 관계 또한 relationship 메소드를 통해 설정 가능하며, cascade 옵션을 통해 업데이트 또는 삭제시 연관 동작을 정의한다.

CRUD 방식

일단 SeesionLocal을 통해 db라는 변수에 Session이 설정되어있다고 가정한다.

db = SessionLocal()

Create(데이터 생성)

생성 과정은 User 객체를 정의하고, .add 명령어를 통해 추가하며, commit()을 통해 반영한다.

refresh메소드를 통해 추가한 객체에 대해서 정보를 바로 로드가 가능하다.

new_user = User( username="legacy_user", email="legacy@example.com", oauth_provider="google", oauth_id="12345" )

db.add(new_user)

db.commit()

db.refresh(new_user) print(new_user.id)

라인 유지용 텍스트

Read(데이터 조회)

조회 명령어는 query 메소드로 시작하며, 연결된 다른 체이닝 호출을 통해 별도 필터링이 가능하다.

하도 코드 조작하는데 라인 계속 무너져서 일단 그냥 평문으로 넣는다.

users = db.query(User).all()

user = db.query(User).filter(User.username == "legacy_user").first()

user = db.query(User).filter_by(email="legacy@example.com").first()

user = db.query(User).get(1)

Update(데이터 수정)

객체를 수정하기 VS 쿼리로 바로 업데이트 하기, 두가지 방식이 존재한다.

안전을 위해서는 객체 로드 후 수정을 하는 방안이 나아보인다.

user = db.query(User).get(1)

if user: user.username = "updated_legacy_user"

db.commit() 라인 유지용 텍스트

조회 없이 바로 접근해서 처리하는 방안은 다음과 같다

update({업데이트할 컬럼: 값})

db.query(User).filter(User.is_active == False).\
    update({User.is_active: True}, synchronize_session=False)

db.commit()

Delete(데이터 삭제)

삭제도 업데이트와 마찬가지로 두가지 방식이 존재한다.

객체 로드 후 바로 삭제하는 방안

user = db.query(User).get(1) if user: db.delete(user) db.commit() 라인 유지용 텍스트

조회 없이 바로 삭제하는 방법

db.query(User).filter(User.username == "legacy_user").delete() db.commit() 라인 유지용 텍스트

2.0 버전

현재 모던 스타일로 2.0 버전이 나와있다고는 하는데,

일단 난 1.4(Legacy 방식)으로 작성중이다.

데이터가 충분히 쌓여있기도 하고,

심화 프로젝트때 디벨롭 하면서 추가로 익힐 수 있지 않을까? 라는 생각을 가지고 있기 때문이다.

큰 차이가 있기보다는 사용하는 문법이나 방식이 조금 현대화 되어있다고 한다.