Python – SQLAlchemy

CRUD 의 시점을 session.commit() 이 호출되는 시점으로 최대한 미루기때문에,
데이타베이스 접속을 최소화할 수 있습니다.

다만, session.commit() 을 누락하면 당연히 롤백됩니다.

테이블 생성

CREATE TABLE `user` (
    `id` INT(11) NOT NULL AUTO_INCREMENT,
    `username` VARCHAR(256) NOT NULL DEFAULT '0' COLLATE 'utf8mb4_general_ci',
    `email` VARCHAR(256) NOT NULL DEFAULT '0' COLLATE 'utf8mb4_general_ci',
    PRIMARY KEY (`id`) USING BTREE
)
COLLATE='utf8mb4_general_ci'
ENGINE=InnoDB
;

VSCode 에서 session 에 대한 자동완성이 필요하면 Pylance 를 설치해야 합니다.
설치 후 VSCode 를 재실행해야 합니다.

from __future__ import annotations

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy import Column, Integer, String, DateTime, Boolean

engine = create_engine('mysql+mysqldb://root:12345678@localhost:3306/testdb')
Session = sessionmaker(engine)
session = Session()

Base = declarative_base()

class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True)
    username = Column(String)
    email = Column(String)
    # time = Column(DateTime, default=datetime.datetime.utcnow, primary_key=True)
    def __init__(self, user: User):
        # self.id = user.id
        self.username = user.username
        self.email = user.email
    def __init__(self, username: str, email: str):
        # self.id = user.id
        self.username = username
        self.email = email
    def update(self, user: User):
        self.username = user.username
        self.email = user.email

테스트 환경 구축

pip install sqlalchemy
pip install mysqlclient

insert

# insert
user = User(username="Lee", email="aaa@aaa.com")
session.add(user)
session.commit()

select

user: User 와 같이 타입을 명시해 주어야 자동완성이 작동합니다.

# select one
user: User = session.query(User).filter(User.username == "Lee").first()
# user: User = session.query(User).filter(User.username == "Lee 2", User.id == 2).first()
print(user.id)

# select all
users: List[User] = session.query(User).filter(User.username == "Lee").all()
# users: List[User] = session.query(User).filter(User.username == "Lee 2").order_by(User.id.desc()).all()
for user in users:
    print(user.id)

update

# update one
user2 = User(username="Lee 2", email="aaa2@aaa.com")
user: User = session.query(User).filter(User.username == "Lee").first()
user.update(user2)

# update list
user2 = User(username="Lee 2", email="aaa2@aaa.com")
users: List[User] = session.query(User).filter(User.username == "Lee").all()
for user in users:
    user.update(user2)

# 커밋을 할 때 자동으로 update 문이 실행된다.
session.commit()

delete

# delete
user: User = session.query(User).filter(User.id == 1).first()
if user is not None:
    session.delete(user)
session.commit()