실습 : ORM 기반 블로그 애플리케이션

사용자에게 블로그 게시글을 작성하고, 조회하고, 수정하고, 삭제할 수 있는 기능을 제공하는 ORM 기반의 애플리케이션을 구현해 보겠습니다. 

 

먼저, models.py 파일을 생성합니다. 해당 파일은 데이터베이스 모델과 SQLAlchemy 세션을 정의하며, 데이터베이스 연결을 설정하고, ORM 모델을 정의하고, 데이터베이스 테이블을 생성하는 역할을 합니다.

 

engine = create_engine('sqlite:///blog.db')

SQLite 데이터베이스 파일을 생성하고 연결을 설정합니다. 실행 결과로 blog.db 파일이 생성됩니다. 

 

Base = declarative_base()

SQLAlchemy ORM 모델의 기초 클래스 역할을 하는 Base 객체를 생성합니다. 

 

class Post(Base):
    __tablename__ = 'posts'

    id = Column(Integer, primary_key=True)
    title = Column(String, nullable=False)
    content = Column(Text, nullable=False)
    author = Column(String, nullable=False)
    created_at = Column(DateTime, default=datetime.datetime.now)

Post 클래스는 데이터베이스의 posts 테이블을 나타냅니다. 각 게시글은 id, title, content, author, created_at 컬럼을 가집니다. 

 

Base.metadata.create_all(engine)

Base.metadata.create_all 메서드를 사용해 데이터베이스에 posts 테이블을 생성합니다. 

 

Session = sessionmaker(bind=engine)

SQLAlchemy 세션을 생성하는 sessionmaker 객체를 만듭니다. 이 객체는 데이터베이스와의 상호작용을 관리합니다. 

 

models.py 전체 코드는 다음과 같습니다. 

from sqlalchemy import create_engine, Column, Integer, String, Text, DateTime
from sqlalchemy.orm import declarative_base
from sqlalchemy.orm import sessionmaker
import datetime

# 데이터베이스 엔진 생성
engine = create_engine('sqlite:///blog.db')

# 기초 선언 생성
Base = declarative_base()

# 블로그 게시글 모델 정의
class Post(Base):
    __tablename__ = 'posts'

    id = Column(Integer, primary_key=True)
    title = Column(String, nullable=False)
    content = Column(Text, nullable=False)
    author = Column(String, nullable=False)
    created_at = Column(DateTime, default=datetime.datetime.now)

# 테이블 생성
Base.metadata.create_all(engine)

# 세션 생성기
Session = sessionmaker(bind=engine)

 

 

CRUD 작섭을 수행하는 함수를 정의한 crud_operations.py 파일을 생성합니다. 해당 파일은 데이터베이스와의 상호작용을 처리하는 역할을 수행합니다. 

 

session = Session()

데이터베이스와의 상호작용을 관리하는 세션을 생성합니다. 

 

def create_post(title, content, author):
    new_post = Post(title=title, content=content, author=author)
    session.add(new_post)
    session.commit()
    return new_post

create_post 함수는 새 게시글을 작성을 처리합니다. 매개변수로 제목(title), 내용(content), 작성자(author)를 받아 새로운 Post 객체를 생성하고 세션에 추가한 후 커밋하여 데이터베이스에 저장하고 생성된 게시글 객체를 반환합니다. 

 

def get_all_posts():
    return session.query(Post).all()

get_all_posts 함수는 데이터베이스에 있는 모든 게시글을 조회합니다. session.query(Post).all()을 사용해 게시글 목록을 리스트 형태로 반환합니다. 

 

def get_post(post_id):
    return session.query(Post).filter(Post.id == post_id).first()

get_post 함수는 특정 ID를 가진 게시글을 조회합니다. 매개변수로 게시글의 ID(post_id)를 받아 해당 ID의 게시글을 반환합니다. 게시글이 없으면 None을 반환합니다. 

 

def update_post(post_id, title=None, content=None):
    post = get_post(post_id)
    if post:
        if title:
            post.title = title
        if content:
            post.content = content
        session.commit()
    return post

update_post 함수는 특정 ID를 가진 게시글을 수정합니다. 매개변수로 게시글의 ID(post_id), 새 제목(title), 새 내용(content)를 받아 해당 게시글을 조회한 후, 제목과 내용을 수정하고 커밋하여 데이터베이스에 반영하고 수정된 게시글 객체를 반환합니다. 

 

def delete_post(post_id):
    post = get_post(post_id)
    if post:
        session.delete(post)
        session.commit()
    return post

delete_post 함수는 특정 ID를 가진 게시글을 삭제합니다. 매개변수로 게시글의 ID(post_id)를 받아 해당 게시글을 조회한 후 세션에서 삭제하고 커밋하여 데이터베이스에서 삭제를 반영하고, 삭제된 게시글 객체를 반환합니다. 

 

session.close()

모든 CRUD 작업이 끝나면 세션을 종료하고 데이터베이스 연결을 정리합니다. 

 

crud_operations.py 전체 코드는 다음과 같습니다. 

from models import Session, Post

# 세션 생성
session = Session()

# CREATE - 새 게시글 작성
def create_post(title, content, author):
    new_post = Post(title=title, content=content, author=author)
    session.add(new_post)
    session.commit()
    return new_post

# READ - 모든 게시글 조회
def get_all_posts():
    return session.query(Post).all()

# READ - 특정 게시글 조회
def get_post(post_id):
    return session.query(Post).filter(Post.id == post_id).first()

# UPDATE - 게시글 수정
def update_post(post_id, title=None, content=None):
    post = get_post(post_id)
    if post:
        if title:
            post.title = title
        if content:
            post.content = content
        session.commit()
    return post

# DELETE - 게시글 삭제
def delete_post(post_id):
    post = get_post(post_id)
    if post:
        session.delete(post)
        session.commit()
    return post

# 세션 종료
session.close()

 

 

사용자 인터페이스를 제공하는 app.py 파일을 정의합니다. 사용자가 CRUD 작업을 선택할 수 있도록 사용자 인터페이스를 제공하고, 사용자 입력을 받아 CRUD 함수를 호출합니다. 

 

def display_menu():
    print("\n--- 블로그 관리 시스템 ---")
    print("N. 새 게시글 작성")
    print("R. 모든 게시글 조회")
    print("V. 특정 게시글 조회")
    print("U. 게시글 수정")
    print("D. 게시글 삭제")
    print("Q. 종료")

사용자가 선택할 수 있는 CRUD 작업 메뉴를 표시합니다. 이 함수는 사용자에게 블로그 관리 시스템의 메뉴를 보여줍니다. 

 

def main():
    while True:
        display_menu()
        choice = input("원하는 작업을 선택하세요: ").strip().upper()
        ...

main 함수는 무한 루프를 사용해 사용자 입력을 반복적으로 처리합니다. strip().upper() 함수를 사용해 입력된 문자열의 공백을 제거하고 대문자로 변환합니다. 

 

if choice == 'N':
    title = input("게시글 제목: ").strip()
    content = input("게시글 내용: ").strip()
    author = input("작성자: ").strip()
    post = create_post(title, content, author)
    print(f"게시글 작성 완료: {post.title}, {post.author}, {post.created_at}")

사용자가 N을 입력하면 create_post 함수를 호출해 새 게시글을 작성합니다. 사용자로부터 제목, 내용, 작성자를 입력받아 새 게시글을 생성하고 데이터베이스에 저장합니다. 

 

elif choice == 'R':
    posts = get_all_posts()
    print("\n모든 게시글:")
    for post in posts:
        print(f"{post.id}: {post.title} by {post.author} at {post.created_at}")

사용자가 R을 입력하면 get_all_posts 함수를 호출해 모든 게시글을 조회합니다. 조회된 게시글 목록을 출력합니다. 

 

elif choice == 'V':
    try:
        post_id = int(input("조회할 게시글 ID: ").strip())
        post = get_post(post_id)
        if post:
            print(f"\n게시글 ID: {post.id}")
            print(f"게시글 제목: {post.title}")
            print(f"게시글 내용: {post.content}")
            print(f"작성자: {post.author}")
            print(f"작성일시: {post.created_at}")
        else:
            print(f"게시글 {post_id}을(를) 찾을 수 없습니다.")
    except ValueError:
        print("잘못된 입력입니다. 게시글 ID는 숫자여야 합니다.")

사용자가 V를 입력하면 특정 게시글을 조회합니다. 사용자로부터 글 ID를 입력받아 get_post 함수를 호출하고, 조회된 게시글 정보를 출력합니다. 입력이 숫자가 아닌 경우 예외를 처리합니다. 

 

elif choice == 'U':
    try:
        post_id = int(input("수정할 게시글 ID: ").strip())
        title = input("새 제목 (건너뛰려면 엔터): ").strip()
        content = input("새 내용 (건너뛰려면 엔터): ").strip()
        post = update_post(post_id, title if title else None, content if content else None)
        if post:
            print(f"게시글 {post_id} 수정 완료: {post.title}, {post.content}")
        else:
            print(f"게시글 {post_id}을(를) 찾을 수 없습니다.")
    except ValueError:
        print("잘못된 입력입니다. 게시글 ID는 숫자여야 합니다.")

사용자가 U를 입력하면 특정 게시글을 수정합니다. 사용자로부터 게시글 ID와 새 제목, 새 내용을 입력받아 update_post 함수를 호출합니다. 게시글이 수정되면 성공 메시지를 출력하고, 게시글을 찾지 못하면 오류 메시지를 출력합니다. 입력이 숫자가 아닌 경우 예외를 처리합니다. 

 

elif choice == 'D':
    try:
        post_id = int(input("삭제할 게시글 ID: ").strip())
        post = delete_post(post_id)
        if post:
            print(f"게시글 {post_id} 삭제 완료.")
        else:
            print(f"게시글 {post_id}을(를) 찾을 수 없습니다.")
    except ValueError:
        print("잘못된 입력입니다. 게시글 ID는 숫자여야 합니다.")

사용자가 D를 입력하면 특정 게시글을 삭제합니다. 사용자로부터 게시글 ID를 입력받아 delete_post 함수를 호출하고, 게시글이 삭제되면 성공 메시지를 출력합니다. 게시글을 찾지 못하면 오류 메시지를 출력합니다. 입력이 숫자가 아닌 경우 예외를 처리합니다. 

 

elif choice == 'Q':
    print("프로그램을 종료합니다.")
    break
    
elif choice == 'Q':
    print("프로그램을 종료합니다.")
    break

사용자가 Q를 입력하면 프로그램을 종료하고, 사용자가 유효하지 않은 선택을 했을 때 경고 메시지를 출력합니다. 

app.py 파일의 전체 코드는 다음과 같습니다. 

from crud_operations import create_post, get_all_posts, get_post, update_post, delete_post

def display_menu():
    print("\n--- 블로그 관리 시스템 ---")
    print("N. 새 게시글 작성")
    print("R. 모든 게시글 조회")
    print("V. 특정 게시글 조회")
    print("U. 게시글 수정")
    print("D. 게시글 삭제")
    print("Q. 종료")

def main():
    while True:
        display_menu()
        choice = input("원하는 작업을 선택하세요: ")

        if choice == 'N':
            title = input("게시글 제목: ")
            content = input("게시글 내용: ")
            author = input("작성자: ")
            post = create_post(title, content, author)
            print(f"게시글 작성 완료: {post.title}, {post.author}, {post.created_at}")

        elif choice == 'R':
            posts = get_all_posts()
            print("\n모든 게시글:")
            for post in posts:
                print(f"{post.id}: {post.title} by {post.author} at {post.created_at}")

        elif choice == 'V':
            post_id = int(input("조회할 게시글 ID: "))
            post = get_post(post_id)
            if post:
                print(f"\n게시글 ID: {post_id}\n게시글 제목: {post.title}\n게시글 내용: {post.content}\n작성자: {post.author}\n작성일시: {post.created_at}")
            else:
                print(f"게시글 {post_id}을(를) 찾을 수 없습니다.")

        elif choice == 'U':
            post_id = int(input("수정할 게시글 ID: "))
            title = input("새 제목 (건너뛰려면 엔터): ")
            content = input("새 내용 (건너뛰려면 엔터): ")
            post = update_post(post_id, title if title else None, content if content else None)
            if post:
                print(f"게시글 {post_id} 수정 완료: {post.title}, {post.content}")
            else:
                print(f"게시글 {post_id}을(를) 찾을 수 없습니다.")

        elif choice == 'D':
            post_id = int(input("삭제할 게시글 ID: "))
            post = delete_post(post_id)
            if post:
                print(f"게시글 {post_id} 삭제 완료.")
            else:
                print(f"게시글 {post_id}을(를) 찾을 수 없습니다.")

        elif choice == 'Q':
            print("프로그램을 종료합니다.")
            break

        else:
            print("잘못된 입력입니다. 다시 시도하세요.")

if __name__ == "__main__":
    main()