pkglog
EN

Python 데이터베이스: SQLite, PostgreSQL 연결과 ORM

작성 · · 15분 읽기 · 수정 중급 튜토리얼

이 글의 핵심

파일이나 딕셔너리에 저장하던 데이터가 늘어나면 조회 조건과 무결성 때문에 데이터베이스가 필요해집니다. commit을 빼먹어 변경이 반영되지 않는 실수, 사용자 목록과 글을 불러올 때 쿼리가 수십 번 나가는 N+1 문제, 연결 해제와 트랜잭션 처리 순서를 예제와 함께 확인합니다.

들어가며

데이터베이스는 데이터를 안전하게 저장하고 관리하는 핵심 기술입니다. REST API 예제처럼 데이터를 파이썬 리스트에 두면 서버를 재시작할 때 사라지고, JSON 파일에 저장하면 두 요청이 동시에 파일을 쓸 때 한쪽 변경이 덮어써집니다. 데이터베이스는 영속성, 동시 접근 제어, “이메일은 중복될 수 없다” 같은 무결성 규칙, 인덱스를 이용한 빠른 조회를 한꺼번에 제공합니다.

이 글에서는 표준 라이브러리 sqlite3로 SQL을 직접 실행해 보고, 같은 작업을 SQLAlchemy ORM으로 옮긴 뒤 Flask와 연결합니다. SQL을 먼저 보는 이유는 ORM이 결국 SQL을 만들어 보내는 도구라서, 뒤에서 다룰 N+1 문제처럼 ORM에서 생기는 성능 문제도 SQL 수준에서 이해해야 풀리기 때문입니다.

sqlite3로 SQLite 쓰기

SQLite 사용

SQLite는 단일 파일에 표 형태로 데이터를 쌓아 두는 서랍장처럼 동작합니다. connect로 파일을 열고, cursor.execute로 SQL 문을 보낸 뒤 commit으로 디스크에 확정합니다. 예제는 사용자 테이블을 만들고 한 행을 넣고 읽는 흐름입니다.

import sqlite3
# 연결
conn = sqlite3.connect('mydb.db')
cursor = conn.cursor()
# 테이블 생성
cursor.execute('''
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        email TEXT UNIQUE,
        age INTEGER
    )
''')
# 데이터 삽입
cursor.execute(
    'INSERT INTO users (name, email, age) VALUES (?, ?, ?)',
    ('철수', '[email protected]', 25)
)
conn.commit()
# 조회
cursor.execute('SELECT * FROM users')
users = cursor.fetchall()
for user in users:
    print(user)
# 연결 종료
conn.close()

INSERT 문에서 값을 문자열에 직접 넣지 않고 ? 자리표시자와 튜플로 넘긴 부분이 가장 중요합니다. f"INSERT ... VALUES ('{name}')"처럼 f-string으로 SQL을 조립하면 이름에 작은따옴표가 들어가는 순간 문법 오류가 나고, 악의적인 입력이면 '; DROP TABLE users; -- 같은 SQL 인젝션이 됩니다. 자리표시자를 쓰면 드라이버가 값을 SQL과 분리해 전달하므로 이 문제가 원천적으로 막힙니다. 값이 하나뿐이어도 ('철수',)처럼 쉼표를 붙여 튜플로 넘겨야 하며, ('철수')는 문자열이라 “Incorrect number of bindings supplied” 오류가 납니다.

commit()을 빠뜨리는 것은 처음 SQLite를 쓸 때 거의 누구나 겪는 실수입니다. 같은 연결에서는 방금 넣은 데이터가 조회되기 때문에 제대로 들어간 것처럼 보이지만, commit() 없이 close()하면 변경이 버려집니다. 스크립트를 다시 실행했더니 데이터가 없다면 이것부터 확인하세요. with conn: 블록을 쓰면 블록이 정상 종료될 때 커밋, 예외가 나면 롤백이 자동으로 됩니다(연결을 닫지는 않으므로 close()는 따로 호출해야 합니다).

SQLite는 파일 하나에 전체 DB가 들어 있어 설치도 서버도 필요 없습니다. 대신 쓰기 잠금이 DB 파일 전체에 걸리므로 여러 프로세스가 동시에 쓰면 “database is locked” 오류가 나기 쉽습니다. 웹 서버 워커 여러 개가 같은 SQLite 파일에 자주 쓰는 구조라면 WAL 모드(PRAGMA journal_mode=WAL)로 완화하거나 PostgreSQL 같은 서버형 DB로 옮기는 것을 검토합니다.

SQLAlchemy ORM 설정과 모델 정의

설치 및 설정

pip install sqlalchemy
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, sessionmaker
# 데이터베이스 연결
engine = create_engine('sqlite:///mydb.db')
Base = declarative_base()
Session = sessionmaker(bind=engine)
session = Session()

세 객체의 역할이 다릅니다. engine은 DB 연결 풀과 SQL 방언(SQLite, PostgreSQL 등)을 관리하고, URL만 postgresql+psycopg://user:pw@host/db로 바꾸면 나머지 코드는 대부분 그대로 쓸 수 있습니다. Base는 모델 클래스들이 상속하는 기반 클래스로, 정의된 테이블 정보(Base.metadata)를 모아 둡니다. session은 객체의 변경 사항을 추적했다가 commit() 때 한 트랜잭션으로 DB에 반영하는 작업 단위입니다.

declarative_base는 SQLAlchemy 2.0부터 sqlalchemy.orm에서 가져옵니다. 예전 경로인 sqlalchemy.ext.declarative는 폐기 예정 경고가 나옵니다. 2.0 스타일에서는 class Base(DeclarativeBase): pass와 Mapped[int] = mapped_column(...) 타입 힌트 방식이 권장되지만, 이 글은 개념에 집중하기 위해 Column 방식으로 설명합니다. 개발 중에는 create_engine(..., echo=True)로 실제 실행되는 SQL을 로그로 보면 ORM이 무엇을 하는지 파악하기 쉽습니다.

모델 정의

class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)
    email = Column(String(100), unique=True)
    age = Column(Integer)
    
    def __repr__(self):
        return f"<User(name='{self.name}', email='{self.email}')>"
# 테이블 생성
Base.metadata.create_all(engine)

모델 클래스의 각 Column이 테이블 컬럼이 되고, String(100)은 VARCHAR(100), unique=True는 유니크 제약으로 바뀝니다. create_all은 없는 테이블만 만들고 이미 있는 테이블은 건드리지 않습니다. 모델에 컬럼을 추가한 뒤 다시 실행해도 기존 테이블에는 컬럼이 생기지 않아 “no such column” 오류가 나는데, 이 때문에 운영 DB의 스키마 변경은 Alembic 같은 마이그레이션 도구로 관리합니다.

SQLite 절의 sqlite3 예제와 같은 mydb.db 파일, 같은 users 테이블 이름을 쓰고 있다는 점도 주의하세요. 테이블이 이미 있으므로 create_all은 아무것도 하지 않고, 두 정의가 다르면 ORM이 기대하는 구조와 실제 테이블이 어긋납니다.

SQLAlchemy로 CRUD

Create (생성)

# 단일 생성
new_user = User(name='철수', email='[email protected]', age=25)
session.add(new_user)
session.commit()
# 여러 개 생성
users = [
    User(name='영희', email='[email protected]', age=30),
    User(name='민수', email='[email protected]', age=28)
]
session.add_all(users)
session.commit()

session.add()는 객체를 세션에 등록할 뿐 바로 INSERT를 보내지 않습니다. 실제 SQL은 commit()이나 조회 직전의 자동 flush 때 나갑니다. 커밋 후에는 new_user.id에 DB가 부여한 기본 키가 채워집니다.

SQLite 절에서 이미 [email protected]을 넣었다면 이 코드는 IntegrityError: UNIQUE constraint failed: users.email로 실패합니다. 이때 세션은 실패한 트랜잭션 상태로 남아서, rollback()을 호출하기 전까지는 다음 작업이 모두 “This Session’s transaction has been rolled back due to a previous exception during flush” 오류를 냅니다. 원래 오류는 사라지고 이 메시지만 반복해서 보이면 앞선 예외를 처리하지 않은 것이 원인입니다.

Read (조회)

# 전체 조회
all_users = session.query(User).all()
for user in all_users:
    print(user.name, user.email)
# 필터링
young_users = session.query(User).filter(User.age < 30).all()
# 단일 조회
user = session.query(User).filter(User.email == '[email protected]').first()
print(user.name)
# 개수
count = session.query(User).count()
print(f"총 {count}명")

filter(User.age < 30)에서 User.age < 30은 파이썬 불리언이 아니라 SQL 조건식 객체입니다. SQLAlchemy가 연산자를 가로채 WHERE users.age < ?로 바꾸기 때문에, and/or 대신 &/|나 and_(), or_()를 써야 합니다. 파이썬의 and로 조건을 묶으면 한쪽 조건만 적용되거나 오류가 납니다.

first()는 결과가 없으면 None을 돌려주므로 바로 user.name에 접근하면 AttributeError: 'NoneType' object has no attribute 'name'이 납니다. 결과가 정확히 하나여야 한다면 one()(없거나 여러 개면 예외), 없을 수도 있다면 one_or_none()이 의도를 더 분명히 드러냅니다. 기본 키로 찾을 때는 session.get(User, 1)이 간단합니다. 참고로 session.query()는 SQLAlchemy 2.0에서 레거시 API로 분류되며, 새 코드는 session.execute(select(User).where(User.age < 30)).scalars().all() 형태를 권장합니다. 동작은 같습니다.

Update (수정)

# 방법 1: 객체 수정
user = session.query(User).filter(User.name == '철수').first()
user.age = 26
session.commit()
# 방법 2: 쿼리로 수정
session.query(User).filter(User.name == '철수').update({'age': 27})
session.commit()

두 방법은 동작 방식이 다릅니다. 방법 1은 객체를 조회해 속성을 바꾸고, 세션이 변경을 감지해 커밋 때 UPDATE를 보냅니다. 객체 하나를 다룰 때 자연스럽고 모델에 정의한 이벤트나 검증 로직이 적용됩니다. 방법 2는 조회 없이 UPDATE ... WHERE 한 문장을 바로 보내므로 수천 행을 한 번에 바꿀 때 훨씬 빠르지만, 이미 세션에 올라와 있는 객체와 DB 값이 어긋날 수 있고 ORM 이벤트를 거치지 않습니다.

이 예제는 name으로 대상을 찾는데, 이름은 유니크가 아니므로 동명이인이 있으면 방법 1은 첫 번째 사람만, 방법 2는 모두 수정합니다. 수정·삭제 대상은 기본 키처럼 유일한 값으로 지정하는 습관이 중요합니다.

Delete (삭제)

# 방법 1: 객체 삭제
user = session.query(User).filter(User.name == '철수').first()
session.delete(user)
session.commit()
# 방법 2: 쿼리로 삭제
session.query(User).filter(User.age < 20).delete()
session.commit()

삭제도 수정과 같은 차이가 있습니다. session.delete(user)는 관계에 설정한 cascade 규칙(예: 사용자를 지우면 글도 지우기)을 ORM이 처리하지만, 쿼리 delete()는 DELETE 문 하나만 보내므로 ORM 수준의 cascade가 적용되지 않습니다. DB에 외래 키 제약이 있으면 자식 행 때문에 삭제가 실패하고, 없으면 주인 없는 글이 남습니다. 앞 예제에서 first()가 None을 돌려준 상태로 session.delete(None)을 호출하면 오류가 나므로 조회 결과를 확인하는 것도 잊지 마세요. 실제 서비스에서는 행을 지우는 대신 deleted_at 컬럼으로 표시만 하는 소프트 삭제를 쓰기도 합니다.

일대다 관계

일대다 관계

from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship
class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    name = Column(String(100))
    posts = relationship('Post', back_populates='author')
class Post(Base):
    __tablename__ = 'posts'
    
    id = Column(Integer, primary_key=True)
    title = Column(String(200))
    content = Column(String)
    user_id = Column(Integer, ForeignKey('users.id'))
    author = relationship('User', back_populates='posts')
# 사용
user = User(name='철수')
post1 = Post(title='첫 포스트', content='내용', author=user)
post2 = Post(title='두 번째', content='내용2', author=user)
session.add_all([user, post1, post2])
session.commit()
# 조회
user = session.query(User).first()
for post in user.posts:
    print(post.title)

외래 키는 “다” 쪽 테이블(posts.user_id)에 둡니다. 한 사용자가 여러 글을 가지므로 글마다 주인의 id를 저장하는 구조입니다. relationship은 DB 컬럼이 아니라 파이썬 객체 사이를 연결하는 속성으로, user.posts로 글 목록을, post.author로 작성자를 꺼낼 수 있게 합니다. back_populates는 두 속성이 같은 관계의 양쪽이라는 것을 알려 주어, post1.author = user로 설정하면 user.posts에도 자동으로 들어갑니다. 그래서 예제에서 user_id를 직접 지정하지 않아도 커밋 시 올바르게 채워집니다.

이 코드를 SQLAlchemy 설정 절과 같은 파일에서 실행하면 Table 'users' is already defined for this MetaData instance 오류가 납니다. 같은 Base에 users 테이블을 두 번 정의했기 때문입니다. 실제로는 그 절의 User에 posts 관계를 추가하는 방식으로 합쳐야 하며, 이미 만들어진 mydb.db에는 posts 테이블만 새로 생깁니다.

user.posts에 처음 접근하는 순간 SELECT ... FROM posts WHERE user_id = ? 쿼리가 추가로 실행됩니다(지연 로딩). 사용자 한 명이면 문제가 없지만, 사용자 100명을 불러와 반복문에서 각자의 posts를 출력하면 목록 쿼리 1번에 글 쿼리 100번이 나가는 N+1 문제가 됩니다. 로컬 SQLite에서는 쿼리 하나가 빨라 티가 나지 않다가, 네트워크 너머의 DB에 붙는 운영 환경에서 페이지가 갑자기 느려지는 식으로 드러나는 경우가 많습니다. 해결 방법은 다음 절에서 다룹니다.

Flask + SQLAlchemy API

from flask import Flask, jsonify, request
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///blog.db'
db = SQLAlchemy(app)
class Post(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200), nullable=False)
    content = db.Column(db.Text)
    
    def to_dict(self):
        return {
            'id': self.id,
            'title': self.title,
            'content': self.content
        }
# 테이블 생성
with app.app_context():
    db.create_all()
@app.route('/api/posts', methods=['GET'])
def get_posts():
    posts = Post.query.all()
    return jsonify([post.to_dict() for post in posts])
@app.route('/api/posts', methods=['POST'])
def create_post():
    data = request.get_json()
    post = Post(title=data['title'], content=data['content'])
    db.session.add(post)
    db.session.commit()
    return jsonify(post.to_dict()), 201
if __name__ == '__main__':
    app.run(debug=True)

Flask-SQLAlchemy는 SQLAlchemy를 Flask 요청 수명 주기에 맞춰 감싼 확장입니다(pip install flask-sqlalchemy). 가장 큰 차이는 세션 관리입니다. 앞 절에서는 session을 직접 만들고 닫아야 했지만, 여기서 db.session은 요청마다 자동으로 준비되고 요청이 끝나면 정리됩니다. 요청 처리 중 예외가 나서 커밋하지 못해도 다음 요청은 깨끗한 세션으로 시작합니다.

db.create_all()을 with app.app_context(): 안에서 호출하는 이유는, Flask-SQLAlchemy 3.0부터 DB 작업에 애플리케이션 컨텍스트가 필요하기 때문입니다. 이 블록 없이 호출하면 “RuntimeError: Working outside of application context.” 오류가 납니다. 요청 처리 함수 안에서는 Flask가 컨텍스트를 자동으로 만들어 주므로 신경 쓸 필요가 없습니다. Post.query.all()도 3.x에서는 레거시 방식이고 db.session.execute(db.select(Post)).scalars().all()이 권장 형태입니다.

create_post는 data['title']이 없으면 KeyError로, title이 null이면 nullable=False 때문에 IntegrityError로 500 오류가 납니다. API 입력 검증과 오류 응답은 REST API에서 다룬 방식으로 보완해야 합니다.

연결 해제·트랜잭션·N+1 쿼리

DB 세션은 쓰고 나면 반드시 정리해야 하는 대여 물건과 비슷합니다. with나 try/except/finally로 연결과 트랜잭션을 묶어 두면, 중간에 오류가 나도 롤백 후 안전하게 닫을 수 있습니다. 관계가 있는 테이블을 한 번에 불러올 때는 N+1(행마다 추가 쿼리)을 의심해 보세요.

# ✅ 연결 관리
with engine.connect() as conn:
    result = conn.execute(query)
# ✅ 트랜잭션
try:
    session.add(user)
    session.commit()
except Exception as e:
    session.rollback()
    raise
# ✅ 쿼리 최적화
# N+1 문제 해결
from sqlalchemy.orm import joinedload
users = session.query(User).options(
    joinedload(User.posts)
).all()

트랜잭션 블록에서 except 후 raise로 예외를 다시 올리는 점이 중요합니다. rollback()만 하고 예외를 삼키면 호출한 쪽은 저장이 성공한 줄 압니다. 롤백은 세션을 다시 쓸 수 있는 상태로 되돌리는 일이고, 실패를 알리는 일은 따로 해야 합니다. SQLAlchemy 2.0에서는 with Session(engine) as session, session.begin(): 형태로 쓰면 정상 종료 시 커밋, 예외 시 롤백, 끝나면 세션 닫기까지 자동으로 처리됩니다.

joinedload는 LEFT OUTER JOIN으로 사용자와 글을 한 번의 쿼리로 가져옵니다. 쿼리 수는 1개로 줄지만, 사용자 한 명에 글이 100개면 사용자 컬럼이 100번 반복된 결과 행을 받게 됩니다. 일대다 관계에서는 selectinload(User.posts)가 보통 더 효율적인데, 사용자 목록을 먼저 조회한 뒤 WHERE user_id IN (...) 쿼리 하나로 모든 글을 가져와 쿼리 2개로 끝납니다. 다대일(Post.author)처럼 부모가 하나인 관계는 joinedload가 잘 맞습니다. 어느 쪽이든 echo=True로 실제 쿼리 수를 확인해 보는 것이 가장 확실합니다.

데이터베이스 요약

  1. SQLite: 파일 기반, 간단한 DB
  2. SQLAlchemy: Python ORM 라이브러리
  3. 모델: 클래스로 테이블 정의
  4. CRUD: Create, Read, Update, Delete
  5. 관계: ForeignKey, relationship

다음 단계

같이 보면 좋은 글

자주 묻는 질문 (FAQ)

Q. 사용자 목록과 각 사용자의 글을 불러올 때 쿼리가 너무 많이 나가는 이유는 무엇인가요?

A. 관계 속성은 기본적으로 처음 접근할 때 따로 조회하기 때문에, 사용자 N명을 불러온 뒤 각자의 posts에 접근하면 1번의 목록 쿼리와 N번의 추가 쿼리가 나가는 N+1 문제가 생깁니다. SQLAlchemy에서는 session.query(User).options(joinedload(User.posts))처럼 관계를 함께 불러오도록 지정해 쿼리 수를 줄입니다. 같은 맥락에서 쓰기 작업은 try로 감싸 실패 시 session.rollback()을 호출해야 세션이 깨진 상태로 남지 않습니다.