[SqlAlchemy] 기록 : session생성(수정일 : 2023/11/10)

 출처 : https://docs.sqlalchemy.org/en/20/orm/quickstart.html

 

중요점 : 기존DB 테이블 참조가능 (이게 의외로 사람들이 바로 설명 안해주었다 내가 못찾은 걸 수도있지만) 

 

 

SqlAlchemy에는 DB를 다루기위해 engine을 만들고 session을 만드는데

session을 만들때 여러가지 방법이 있다. 

그중 scoped_session을 사용하였다. 

그냥 sessionmaker와 다른점은 인스턴스화되어 몇번 불려도 같은 session이 불린다는것이 다른점이다.

 

아래는 qiita에서 찾은 아주 기본방식의 사용법이다. 

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

engine = create_engine(～略～)

session = scoped_session(
            sessionmaker(
                autocommit = False,
                autoflush = True,
                bind = engine))

Base = declarative_base()
# 予めテーブル定義の継承元クラスにqueryプロパティを仕込んでおく
Base.query = session.query_property()

 

 

from sqlalchemy import Column, Integer, String
from settings import Base

# UserクラスはBaseを継承しているのでqueryプロパティを持つ
# XXX: ここで直接queryプロパティを仕込んでも良い
class User(Base):
    __tablename__ = 'user_list'
    id = Column('id', Integer, primary_key=True)
    name = Column('name', String(100))

 

 

from settings import session
from model import User

# Seesionを使ってるように見えないが内部的に使ってる
result = User.query.filter_by(id == 1).all()

# 特にcommitに意味はないけどサンプルコードとして
session.commit()

 

 

위와 같이 Base로 만들지않고 engine과 session을 생성하는 setup.py나 sqlprovider.py파일안에서

class화해서 session을 def로 메소드로 만들어 orm을사용하려는 곳에서 import로 가져와서 사용해도된다.

 

다음은 sqlAlchemy엔진을 만들고나서 연결할 DB에 존재하거나 존재할 테이블의 구조에대해 선언하는곳이다.

>>> from typing import List
>>> from typing import Optional
>>> from sqlalchemy import ForeignKey
>>> from sqlalchemy import String
>>> from sqlalchemy.orm import DeclarativeBase
>>> from sqlalchemy.orm import Mapped
>>> from sqlalchemy.orm import mapped_column
>>> from sqlalchemy.orm import relationship

>>> class Base(DeclarativeBase):
...     pass

>>> class User(Base):
...     __tablename__ = "user_account"
...
...     id: Mapped[int] = mapped_column(primary_key=True)
...     name: Mapped[str] = mapped_column(String(30))
...     fullname: Mapped[Optional[str]]
...
...     addresses: Mapped[List["Address"]] = relationship(
...         back_populates="user", cascade="all, delete-orphan"
...     )
...
...     def __repr__(self) -> str:
...         return f"User(id={self.id!r}, name={self.name!r}, fullname={self.fullname!r})"

>>> class Address(Base):
...     __tablename__ = "address"
...
...     id: Mapped[int] = mapped_column(primary_key=True)
...     email_address: Mapped[str]
...     user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))
...
...     user: Mapped["User"] = relationship(back_populates="addresses")
...
...     def __repr__(self) -> str:
...         return f"Address(id={self.id!r}, email_address={self.email_address!r})"

이렇게하면 sqlalchemy조작법을 검색했을때 흔히나오는 테이블 생성부터

안해도되고 기존에 만들어진 DB의 테이블을 참조가능하다.

 

Select예제 

>>> from sqlalchemy import select

>>> session = Session(engine)

>>> stmt = select(User).where(User.name.in_(["spongebob", "sandy"]))

>>> for user in session.scalars(stmt):
...     print(user)
BEGIN (implicit)
SELECT user_account.id, user_account.name, user_account.fullname
FROM user_account
WHERE user_account.name IN (?, ?)
[...] ('spongebob', 'sandy')
User(id=1, name='spongebob', fullname='Spongebob Squarepants')
User(id=2, name='sandy', fullname='Sandy Cheeks')

stmt = select(User).where(User.name.in_(["spongebob", "sandy"]))
이부분을 보면 흔히 사용하는 orm방식이다.

 

 

실행을보면 excute등이 아니고 scalars라는게 있는데 
공식 도큐멘트에서는 

 

from sqlalchemy import select
from sqlalchemy.orm import Session

with Session(engine) as session:
    # query for ``User`` objects
    statement = select(User).filter_by(name="ed")

    # list of ``User`` objects
    user_obj = session.scalars(statement).all()

    # query for individual columns
    statement = select(User.name, User.fullname)

    # list of Row objects
    rows = session.execute(statement).all()

이렇게 나와있다.

 

근데 sqlAlchemy의 orm가이드를 보면 보통 

result = session.execute(stmt)

이렇게 사용하고  리스트로 데이터가 들어오면 for등을 이용해 result를 하나하나 분리해서 다루는것 같다.