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

들어가며

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

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

---

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

---

정리

핵심 요약

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

다음 단계

• 배포
• Pandas 데이터 분석

---

관련 글

• Django 기초 | Python 풀스택 웹 프레임워크 시작하기