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 세션은 쓰고 나면 반드시 정리해야 하는 대여 물건과 비슷합니다. 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 문제 해결
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) 수정 후 회귀·부하 테스트.
정리
핵심 요약
- SQLite: 파일 기반, 간단한 DB
- SQLAlchemy: Python ORM 라이브러리
- 모델: 클래스로 테이블 정의
- CRUD: Create, Read, Update, Delete
- 관계: ForeignKey, relationship
다음 단계
관련 글
자주 묻는 질문 (FAQ)
Q. 이 내용을 실무에서 언제 쓰나요?
A. Python 데이터베이스에 대해 정리한 개발 블로그 글입니다. conn = sqlite3.connect( 데이터베이스는 데이터를 안전하게 저장하고 관리하는 핵심 기술입니다. --- import sqlite3. Sta… 실무에서는 위 본문의 예제와 선택 가이드를 참고해 적용하면 됩니다.
Q. 선행으로 읽으면 좋은 글은?
A. 각 글 하단의 이전 글 또는 관련 글 링크를 따라가면 순서대로 배울 수 있습니다. Python 시리즈 목차에서 전체 흐름을 확인할 수 있습니다.
Q. 더 깊이 공부하려면?
A. cppreference와 해당 라이브러리 공식 문서를 참고하세요. 글 말미의 참고 자료 링크도 활용하면 좋습니다.
같이 보면 좋은 글 (내부 링크)
이 주제와 연결되는 다른 글입니다.
- Python REST API | Flask/Django로 API 서버 만들기
- Python 데코레이터 | @decorator 완벽 정리
- Pandas 기초 | Python 데이터 분석 라이브러리 완벽 정리
이 글에서 다루는 키워드 (관련 검색어)
Python, 데이터베이스, Database, SQLite, PostgreSQL, SQLAlchemy, ORM 등으로 검색하시면 이 글이 도움이 됩니다.