[DB] SQLAlchemy

집중맞은 도둑력·2024년 8월 17일

DB

목록 보기

3/3

0. 🔖 목차

SQLAlchemy 개요 및 시작
DDL 매핑
DML 매핑

1. SQLAlchemy 개요 및 시작

파이썬에서 데이터베이스 작업을 쉽게 하기 위해 제공되는 라이브러리.

ORM 기능을 통해 파이썬 객체와 데이터베이스 테이블 간의 매핑을 간편하게 해줌.

1-1. SQLAlchemy 설치

파이썬 라이브러리이기 때문에 아래와 같이 설치를 해야한다.

pip install sqlalchemy

1-2. 주요 구성 요소

Engine: 데이터베이스와의 연결을 표현함. 데이터베이스에 대한 메타데이터를 관리하며 실제 SQL 쿼리를 실행
Session: 데이터베이스와의 대화를 관리함. ORM의 주요 인터페이스, DML시 사용
Declarative Base: 테이블과 클래스를 매핑하는 데 사용되는 기본 클래스, DDL시 사용
Mapped Class: 실제로 테이블과 매핑되는 클래스

1-3. SQLAlchemy 시작

1-3-1. 데이터베이스 엔진 설정

메모리 데이터베이스와 연결
데이터베이스와 연결할 엔진을 설정한다.

from sqlalchemy import create_engine

engine = create_engine('sqlite:///:memory:', echo=True)

호스팅 데이터베이스와 연결

데이터베이스 드라이버 설치
데이터베이스에 연결하기 위해 각 데이터베이스에 맞는 드라이버 설치

## MySQL
pip install pymysql 

## PostgreSQL
pip install psycopg2

데이터베이스 연결 URL 지정
SQLAlchemy는 DB와 연결할 때 아래 형태의 URL을 통해 연결함

## MySQL
DATABASE_URL = "mysql+<드라이버>://<유저명>:<비밀번호>@<호스트>/<DB명>"
DATABASE_URL = "mysql+pymysql://user:1234@localhost/mydb" #ex

## PostgreSQL
DATABASE_URL = "postgresql+<드라이버>://<유저명>:<비밀번호>@<호스트>/<DB명>"
DATABASE_URL = "postgresql+psycopg2://user:1234@localhost/mydb" #ex

데이터베이스 연결

from sqlalchemy import create_engine

engine = create_engine(DATABASE_URL, echo=True)

1-3-2. Declarative Base 생성

테이블과 매핑할 클래스를 만들기 위해 Declarative Base를 생성한다.

from sqlalchemy.ext.declarative import DeclarativeBase
from sqlalchemy.orm import registry
from sqlalchemy import Column, Integer, String, DateTime, func
import uuid
from sqlalchemy.dialects.mysql import CHAR, VARCHAR

class Base(DeclarativeBase):
    registry = registry(
        type_annotation_map={
            str: VARCHAR(255),  # 기본적으로 모든 str 타입을 VARCHAR(255)로 매핑
            uuid.UUID: CHAR(36)  # UUID를 CHAR(36)으로 매핑 (MySQL에서 UUID 표현)
        }
    )

    id = Column(Integer, primary_key=True, autoincrement=True)
    created_at = Column(DateTime, default=func.now())
    updated_at = Column(DateTime, default=func.now(), onupdate=func.now())

    def save(self, session):
        session.add(self)
        session.commit()

    def delete(self, session):
        session.delete(self)
        session.commit()

    def __repr__(self):
        return f"<{self.__class__.__name__}(id={self.id})>"

2. DDL 매핑

SQLAlchemy는 파이썬 코드를 사용하여 데이터베이스 스키마(즉, 테이블, 열, 제약 조건 등)를 정의하고 조작할 수 있다.

from sqlalchemy import (Table, 
					select,
					Column, 
					Integer, 
					String, 
					create_engine, 
					ForeignKey, 
					Index, 
					Numeric, 
					DateTime, 
					func,
                    event,
                    MetaData)
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

아래 예제에서 ORM 부분은 위 코드가 항상 포함되어있다고 가정

2-1. 간단한 테이블 생성

SQL

CREATE TABLE users (
    id INT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(255),
    email VARCHAR(255) UNIQUE
);

SQLAlchemy

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(255), nullable=False)
    email = Column(String(255), unique=True, nullable=False)

2-2. 기본키와 외래 키 관계

SQL

CREATE TABLE departments (
    id INT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(255) UNIQUE
);

CREATE TABLE employees (
    id INT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(255),
    department_id INT,
    FOREIGN KEY (department_id) REFERENCES departments(id)
);

SQLAlchemy

class Department(Base):
    __tablename__ = 'departments'

    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(255), unique=True, nullable=False)

    employees = relationship('Employee', back_populates='department')

class Employee(Base):
    __tablename__ = 'employees'

    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(255), nullable=False)
    department_id = Column(Integer, ForeignKey('departments.id'))

    department = relationship('Department', back_populates='employees')

2-3. 다대다 관계

SQL

CREATE TABLE students (
    id INT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(255)
);

CREATE TABLE courses (
    id INT AUTO_INCREMENT PRIMARY KEY,
    title VARCHAR(255)
);

CREATE TABLE student_courses (
    student_id INT,
    course_id INT,
    PRIMARY KEY(student_id, course_id),
    FOREIGN KEY (student_id) REFERENCES students(id),
    FOREIGN KEY (course_id) REFERENCES courses(id)
);

SQLAlchemy

student_courses = Table('student_courses', Base.metadata,
    Column('student_id', Integer, ForeignKey('students.id'), primary_key=True),
    Column('course_id', Integer, ForeignKey('courses.id'), primary_key=True)
)

class Student(Base):
    __tablename__ = 'students'

    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(255), nullable=False)

    courses = relationship('Course', secondary=student_courses, back_populates='students')

class Course(Base):
    __tablename__ = 'courses'

    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(255), nullable=False)

    students = relationship('Student', secondary=student_courses, back_populates='courses')

2-4. 고급 DDL: 인덱스, 제약 조건, 기본값 등

SQL

CREATE TABLE products (
    id INT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(255) NOT NULL,
    price DECIMAL(10, 2) NOT NULL DEFAULT 0.00,
    stock INT NOT NULL DEFAULT 0,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    UNIQUE (name),
    INDEX idx_price (price)
);

SQLAlchemy

class Product(Base):
    __tablename__ = 'products'

    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(255), unique=True, nullable=False)
    price = Column(Numeric(10, 2), nullable=False, default=0.00)
    stock = Column(Integer, nullable=False, default=0)
    created_at = Column(DateTime, server_default=func.now())

    __table_args__ = (
        Index('idx_price', price),
    )

2-5. 트리거와 SQLAlchemy Event Listeners

SQL

CREATE TRIGGER update_stock_before_insert
BEFORE INSERT ON products
FOR EACH ROW
SET NEW.stock = IFNULL(NEW.stock, 0);

SQLAlchemy

@event.listens_for(Product, "before_insert")
def set_default_stock(mapper, connection, target):
    if target.stock is None:
        target.stock = 0

2-6. View 생성 및 관리

SQL

CREATE VIEW product_view AS
SELECT id, name, price FROM products WHERE stock > 0;

SQLAlchemy

product_view = Table('product_view', Base.metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String(255)),
    Column('price', Numeric(10, 2)),
    autoload_with=engine,
    extend_existing=True
)

Base.metadata.create_all(engine)

# 명시적으로 뷰 생성 쿼리 실행
engine.execute("""
    CREATE VIEW product_view AS
    SELECT id, name, price FROM products WHERE stock > 0;
""")

2-7. 파티셔닝, 샤딩, 클러스터링 등 고급 기술

SQL

CREATE TABLE log (
    id INT AUTO_INCREMENT PRIMARY KEY,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
PARTITION BY RANGE (YEAR(created_at)) (
    PARTITION p0 VALUES LESS THAN (2021),
    PARTITION p1 VALUES LESS THAN (2022),
    PARTITION p2 VALUES LESS THAN (2023)
);

SQLAlchemy

# 수동으로 파티셔닝 테이블 생성
metadata = MetaData()

log = Table('log', metadata,
    Column('id', Integer, primary_key=True),
    Column('created_at', DateTime, server_default=func.now()),
    mysql_partition_by='RANGE (YEAR(created_at))',
    mysql_partitions=[
        'PARTITION p0 VALUES LESS THAN (2021)',
        'PARTITION p1 VALUES LESS THAN (2022)',
        'PARTITION p2 VALUES LESS THAN (2023)'
    ]
)

metadata.create_all(engine)

3. DML 매핑

SQLAlchemy는 파이썬 객체를 통해 데이터베이스에서 데이터를 삽입, 조회, 수정, 삭제할 수 있는 기능을 제공한다

# 세션 생성
session = db.get_session()

...

# 세션 종료
session.close()

아래 예제에서 ORM 부분은 위 코드가 항상 데코되어있다고 가정

3-1. INSERT

SQL

INSERT INTO users (name, email) VALUES ('Alice', 'alice@example.com');

SQLAlchemy

new_user = User(name='Alice', email='alice@example.com')
session.add(new_user)
session.commit()

3-2. SELECT

SQL

SELECT * FROM users WHERE name = 'Alice';

SQLAlchemy

user = session.query(User).filter_by(name='Alice').first()
print(user)

3-3. UPDATE

SQL

UPDATE users SET email = 'newalice@example.com' WHERE name = 'Alice';

SQLAlchemy

user = session.query(User).filter_by(name='Alice').first()
user.email = 'newalice@example.com'
session.commit()

3-4. DELETE

SQL

DELETE FROM users WHERE name = 'Alice';

SQLAlchemy

user = session.query(User).filter_by(name='Alice').first()
session.delete(user)
session.commit()

3-5. JOIN, WHERE, ORDER BY

SQL

SELECT employees.name, departments.name
FROM employees
JOIN departments ON employees.department_id = departments.id
WHERE departments.name = 'Engineering'
ORDER BY employees.name;

SQLAlchemy

results = session.query(Employee.name, Department.name)\
    .join(Department, Employee.department_id == Department.id)\
    .filter(Department.name == 'Engineering')\
    .order_by(Employee.name).all()

for employee_name, department_name in results:
    print(f"{employee_name} works in {department_name}")

3-6. Aggregation and Group By

SQL

SELECT department_id, COUNT(*)
FROM employees
GROUP BY department_id;

SQLAlchemy

results = session.query(Employee.department_id, func.count(Employee.id))\
    .group_by(Employee.department_id).all()

for department_id, count in results:
    print(f"Department {department_id} has {count} employees")

3-7. Subquery

SQL

SELECT name FROM users WHERE id IN (SELECT user_id FROM orders WHERE total > 100);

SQLAlchemy

subq = session.query(Order.user_id).filter(Order.total > 100).subquery()

users = session.query(User.name).filter(User.id.in_(subq)).all()

for user in users:
    print(user.name)

3-8. Bulk Operations

SQL

INSERT INTO users (name, email) VALUES ('Bob', 'bob@example.com'), ('Charlie', 'charlie@example.com');

SQLAlchemy

session.bulk_save_objects([
    User(name='Bob', email='bob@example.com'),
    User(name='Charlie', email='charlie@example.com')
])
session.commit()

3-9. 트랜잭션 처리

SQL

START TRANSACTION;

INSERT INTO users (name, email) VALUES ('Dave', 'dave@example.com');
UPDATE accounts SET balance = balance - 100 WHERE user_id = 1;
UPDATE accounts SET balance = balance + 100 WHERE user_id = 2;

COMMIT;

SQLAlchemy

try:
    new_user = User(name='Dave', email='dave@example.com')
    session.add(new_user)

    account1 = session.query(Account).filter_by(user_id=1).first()
    account2 = session.query(Account).filter_by(user_id=2).first()

    account1.balance -= 100
    account2.balance += 100

    session.commit()
except:
    session.rollback()
    raise

3-10. 이벤트 리스너를 통한 커스텀 DML 처리

SQL에는 없고 SQLAlchemy에서 제공하는 이벤트 리스너 기능을 통해 특정 이벤트에 반응하여 커스텀 동작을 수행할 수 있음

SQLAlchemy

from sqlalchemy import event

@event.listens_for(User, "before_insert")
def generate_user_uuid(mapper, connection, target):
    target.user_uuid = str(uuid.uuid4())

이벤트 리스너 종류
before_insert: 새로운 레코드가 데이터베이스에 삽입되기 전
after_insert: 새로운 레코드가 데이터베이스에 삽입된 후
before_update: 기존 레코드가 업데이트되기 전
after_update: 기존 레코드가 업데이트된 후
before_delete: 레코드가 삭제되기 전
after_delete: 레코드가 삭제된 후

집중맞은 도둑력

틀린_내용이_있다면_말해주세요.

이전 포스트

[DB] SQLAlchemy

DB

0. 🔖 목차

1. SQLAlchemy 개요 및 시작

1-1. SQLAlchemy 설치

1-2. 주요 구성 요소

1-3. SQLAlchemy 시작

1-3-1. 데이터베이스 엔진 설정

1-3-2. Declarative Base 생성

2. DDL 매핑

2-1. 간단한 테이블 생성

2-2. 기본키와 외래 키 관계

2-3. 다대다 관계

2-4. 고급 DDL: 인덱스, 제약 조건, 기본값 등

2-5. 트리거와 SQLAlchemy Event Listeners

2-6. View 생성 및 관리

2-7. 파티셔닝, 샤딩, 클러스터링 등 고급 기술

3. DML 매핑

3-1. INSERT

3-2. SELECT

3-3. UPDATE

3-4. DELETE

3-5. JOIN, WHERE, ORDER BY

3-6. Aggregation and Group By

3-7. Subquery

3-8. Bulk Operations

3-9. 트랜잭션 처리

3-10. 이벤트 리스너를 통한 커스텀 DML 처리

[DB] DB 고급 정리(진행중)

0개의 댓글