본문으로 건너뛰기
Previous
Next
Python 데이터베이스 | SQLite, PostgreSQL, ORM 완벽 정리

Python 데이터베이스 | SQLite, PostgreSQL, ORM 완벽 정리

Python 데이터베이스 | SQLite, PostgreSQL, ORM 완벽 정리

이 글의 핵심

conn = sqlite3.connect( 데이터베이스는 데이터를 안전하게 저장하고 관리하는 핵심 기술입니다. --- import sqlite3'mydb.db') cursor = conn.cursor() 개념과 예제 코드를 단계적으로 다루며, 실무·학습에 참고할 수 있도록 구성했습니다.

들어가며

”데이터를 영구적으로 저장하기”

데이터베이스는 데이터를 안전하게 저장하고 관리하는 핵심 기술입니다.

실무 활용 사례: 데이터 분석, 웹 개발, 자동화 프로젝트에서 실제로 사용한 패턴과 코드를 바탕으로 정리했습니다. 초보자가 흔히 겪는 오류와 해결법을 포함합니다.

실무에서 느낀 Python의 매력

처음 Python을 배울 때는 “이게 정말 프로그래밍 언어인가?” 싶을 정도로 간결했습니다. C++에서 10줄로 작성하던 코드가 Python에서는 2~3줄로 끝나는 경우가 많았죠. 특히 데이터 분석 프로젝트를 진행하면서 Pandas와 NumPy의 강력함을 체감했습니다. 엑셀로 몇 시간 걸리던 작업이 Python 스크립트로는 몇 초 만에 끝나는 걸 보고 동료들이 놀라워했던 기억이 납니다. 하지만 처음부터 순탄하지만은 않았습니다. 들여쓰기 하나 잘못해서 몇 시간을 헤맨 적도 있고, 가상환경 설정이 꼬여서 프로젝트 전체를 다시 시작한 적도 있습니다. 이런 시행착오를 겪으며 깨달은 건, 환경 설정을 처음부터 제대로 하는 것이 얼마나 중요한지였습니다. 이 글에서는 제가 겪은 실수들을 바탕으로, 여러분이 같은 시행착오를 겪지 않도록 실전 팁을 담았습니다.

1. 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()

2. SQLAlchemy ORM

설치 및 설정

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

모델 정의

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)

3. 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()

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}명")

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()

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()

4. 관계 (Relationship)

일대다 관계

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)

5. 실전 예제

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)

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

DB 세션은 쓰고 나면 반드시 정리해야 하는 대여 물건과 비슷합니다. withtry/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 문제 해결
users = session.query(User).options(
    joinedload(User.posts)
).all()

내부 동작과 핵심 메커니즘

이 글의 주제는 「Python 데이터베이스 | SQLite, PostgreSQL, ORM 완벽 정리」입니다. 여기서는 앞선 설명을 구현·런타임 관점에서 한 번 더 압축합니다. 데이터 흐름과 실패 모드를 기준으로 생각하면, “입력이 어디서 검증되고, 핵심 연산이 어디서 일어나며, 부작용(I/O·네트워크·디스크)이 어디서 터지는가”가 한눈에 드러납니다.

처리 파이프라인(개념도)

flowchart TD
  A[입력·요청·이벤트] --> B[파싱·검증·디코딩]
  B --> C[핵심 연산·상태 전이]
  C --> D[부작용: I/O·네트워크·동시성]
  D --> E[결과·관측·저장]

알고리즘·프로토콜 관점에서의 체크포인트

  • 불변 조건(Invariant): 각 단계가 만족해야 하는 조건(예: 버퍼 경계, 프로토콜 상태, 트랜잭션 격리)을 문장으로 적어 두면 디버깅 비용이 줄어듭니다.
  • 결정성: 동일 입력에 동일 출력이 보장되는 순수한 층과, 시간·네트워크에 의해 달라질 수 있는 층을 분리해야 테스트와 장애 분석이 쉬워집니다.
  • 경계 비용: 직렬화/역직렬화, 문자 인코딩, syscall 횟수, 락 경합처럼 “한 번의 호출이 아니라 누적되는 비용”을 의심 목록에 넣습니다.

프로덕션 운영 패턴

실서비스에서는 기능 구현과 함께 관측·배포·보안·비용이 동시에 요구됩니다. 아래는 팀에서 자주 쓰는 최소 체크리스트입니다.

영역운영 관점에서의 질문
관측성요청 단위 상관 ID, 에러율/지연 분위수, 주요 의존성 타임아웃이 보이는가
안전성입력 검증·권한·비밀 관리가 코드 경로마다 일관적인가
신뢰성재시도는 멱등한 연산에만 적용되는가, 서킷 브레이커·백오프가 있는가
성능캐시 계층·배치 크기·풀링·백프레셔가 데이터 규모에 맞는가
배포롤백 룬북, 카나리, 마이그레이션 호환성이 문서화되어 있는가

운영 환경에서는 “개발자 PC에서는 재현되지 않던 문제”가 시간·부하·데이터 크기 때문에 드러납니다. 따라서 스테이징의 데이터 양·네트워크 지연을 가능한 한 현실에 가깝게 맞추는 것이 중요합니다.


문제 해결(Troubleshooting)

증상가능 원인조치
간헐적 실패레이스 컨디션, 타임아웃, 외부 의존성 불안정최소 재현 스크립트 작성, 분산 트레이스·로그 상관관계 확인
성능 저하N+1 쿼리, 동기 I/O, 잠금 경합, 과도한 직렬화프로파일러·APM으로 핫스팟 확인 후 한 가지씩 제거
메모리 증가캐시 무제한, 클로저/이벤트 구독 누수, 대용량 객체의 불필요한 복사상한·TTL·스냅샷 비교(힙 덤프/트레이스)
빌드·배포만 실패환경 변수·권한·플랫폼 차이CI 로그와 로컬 diff, 컨테이너/런타임 버전 핀(pin)

권장 디버깅 순서: (1) 최소 재현 만들기 (2) 최근 변경 범위 좁히기 (3) 의존성·환경 변수 차이 확인 (4) 관측 데이터로 가설 검증 (5) 수정 후 회귀·부하 테스트.

정리

핵심 요약

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

다음 단계


관련 글


자주 묻는 질문 (FAQ)

Q. 이 내용을 실무에서 언제 쓰나요?

A. Python 데이터베이스에 대해 정리한 개발 블로그 글입니다. conn = sqlite3.connect( 데이터베이스는 데이터를 안전하게 저장하고 관리하는 핵심 기술입니다. --- import sqlite3. Sta… 실무에서는 위 본문의 예제와 선택 가이드를 참고해 적용하면 됩니다.

Q. 선행으로 읽으면 좋은 글은?

A. 각 글 하단의 이전 글 또는 관련 글 링크를 따라가면 순서대로 배울 수 있습니다. Python 시리즈 목차에서 전체 흐름을 확인할 수 있습니다.

Q. 더 깊이 공부하려면?

A. cppreference와 해당 라이브러리 공식 문서를 참고하세요. 글 말미의 참고 자료 링크도 활용하면 좋습니다.


같이 보면 좋은 글 (내부 링크)

이 주제와 연결되는 다른 글입니다.


이 글에서 다루는 키워드 (관련 검색어)

Python, 데이터베이스, Database, SQLite, PostgreSQL, SQLAlchemy, ORM 등으로 검색하시면 이 글이 도움이 됩니다.