Python으로 데이터베이스 연결하기

PostgreSQL과 연결하기

데이터베이스 연결

참고: jupyter notebook에서 %를 앞에 붙이면 cmd창에서 해당 명령어를 실행하겠다는 의미

import psycopg2

# 데이터베이스 연결 설정
conn = psycopg2.connect(
    host="localhost", #데이터베이스 호스트 이름
    port=5432,
    database="postgres", #데이터베이스 이름
    user="postgres", #데이터베이스 사용자 이름
    password="123" #데이터베이스 비밀번호
)

# 커서 생성
cur = conn.cursor()

데이터 조회

cur.execute("SELECT * FROM user_info")
rows = cur.fetchall()

for row in rows:
    print(row)

Insert

query = "INSERT INTO user_info VALUES ('chris2', '김객곽', '9303132984027', 'SK', '01048484848');"
cur.execute(query)
conn.commit()

에러발생시

conn.rollback()

Update

query = "UPDATE user_info SET user_name = '김곽객' WHERE user_name = '김객곽';"
try:
    cur.execute(query)
    conn.commit()
except Exception as e:
    conn.rollback()
    print(f"Error: {e}")

DB 연결 해제

cur.close()
conn.close()

안전한 문법

query = "INSERT INTO user_info VALUES (%s, %s, %s, %s, %s);"
cur.execute(query, ('chris4', '곽객객', '9510132222222', 'KT', '01055555555'))

PostgreSQL에 데이터 관리하기

데이터 가져오기

• pandas 이용

cur.execute("SELECT * FROM fms.chick_info")
results = cur.fetchall()
col_names = [desc[0] for desc in cur.description]
df = pd.DataFrame(results, columns=col_names)

# warning 발생
total_result = pd.read_sql_query("SELECT * FROM fms.total_result", conn)

데이터 시각화

데이터 가져오기

import pandas as pd
query = "SELECT c.code_desc, b.raw_weight FROM fms.chick_info a, fms.prod_result b, fms.master_code c WHERE a.chick_no = b.chick_no and a.breeds = c.code"
try:
    cur.execute(query)
except Exception as e:
    conn.rollback()
    print(f"Error: {e}")
results = cur.fetchall()
col_names = [desc[0] for desc in cur.description]
df = pd.DataFrame(results, columns=col_names)

시각화

plt.figure(figsize=(8,4))
sns.boxplot(
    x='code_desc',
    y='raw_weight',
    data=df
)

plt.xlabel('품종')
plt.ylabel('무게')
plt.title('품종별 무게 분포')

plt.show()

---

애플리케이션 개발

Python 기반 Web Framework

Flask

• 파이썬 웹 프레임워크
• 마이크로 웹 프레임워크
• 확장성 있는 설계

기본

from flask import Flask
app = Flask(__name__)

@app.route("/")
def hello():
	return "Hello World!"
    
if __name__ == "__main__":
	app.run()

• 플라스크에서 프로젝트는 하나의 웹 사이트
• 플라스크 프로젝트를 생성 == 웹사이트 1개 생성
• 플라스크 프로젝트 안에는 보통 한 개의 플라스크 애플리케이션이 존재

설치

pip install flask

실행

flask run

Routing

• 어떤 데이터를 목적지까지 보내기 위한 경로를 선정하는 과정
• 라우팅 프로토콜: 라우팅을 할 때 어떻게 데이터가 목적지까지 도달할 길을 만들지를 정의한 규칙

.env

• 공백X

DB_NAME='DB이름'
DB_USER='유저이름'
DB_HOST='localhost'
DB_PORT=5432
DB_PASSWORD=비밀번호

pip install python-dotenv
from dotenv import load_dotenv

app.py

import os
import psycopg2
from psycopg2.extras import DictCursor
from flask import Flask, render_template, request, redirect, url_for, flash, session, jsonify
from dotenv import load_dotenv
from datetime import datetime
import json

# 로컬 환경에서는 .env를 읽고, Azure에서는 패스.
if os.path.exists('.env'):
    load_dotenv()
app = Flask(__name__)
app.secret_key = os.urandom(24)

# 데이터베이스 연결 함수
def get_db_connection():
    conn = psycopg2.connect(
        host=os.getenv('DB_HOST'),
        port=os.getenv('DB_PORT'),
        dbname=os.getenv('DB_NAME'),
        user=os.getenv('DB_USER'),
        password=os.getenv('DB_PASSWORD'),
        sslmode='require' #Azure를 위해 반드시 추가
    )
    print('get_db_connection', conn)
    conn.autocommit = True
    return conn

@app.route('/')
def index():
    # 1. 데이터 베이스에 접속
    conn = get_db_connection()
    print('get_db_connection', conn)
    cursor = conn.cursor(cursor_factory=DictCursor)
    # 2. SELECT
    cursor.execute("SELECT id, title, author, created_at, view_count, like_count FROM board.posts ORDER BY created_at DESC")
    posts = cursor.fetchall()
    cursor.close()
    conn.close()
    # 3. index.html 파일에 변수로 넘겨주기
    return render_template('index.html', posts = posts)

@app.route('/create/', methods=['GET'] )
def create_form():
    return render_template('create.html')

@app.route('/create/',methods=['POST']  )
def create_post():
    #1. 폼에 있는 정보들을 get
    title = request.form.get('title')
    author = request.form.get('author')
    content = request.form.get('content')

    if not title or not author or not content:
        flash('모든 필드를 똑바로 채워주세요!!!!')
        return redirect(url_for('create_form'))
    
    # 1. 데이터 베이스에 접속
    conn = get_db_connection()
    cursor = conn.cursor(cursor_factory=DictCursor)
    # 2. INSERT
    cursor.execute("INSERT INTO board.posts (title, content, author) VALUES (%s, %s, %s) RETURNING id", (title,author,content ))
    post_id = cursor.fetchone()[0]
    cursor.close()
    conn.close()
    flash('게시글이 성공적으로 등록되었음')
    return redirect(url_for('view_post', post_id=post_id))

@app.route('/post/<int:post_id>')
def view_post(post_id):
    conn = get_db_connection()
    cursor = conn.cursor(cursor_factory=DictCursor)
    
    cursor.execute('UPDATE board.posts SET view_count = view_count + 1 WHERE id = %s', (post_id,))
    
    cursor.execute('SELECT * FROM board.posts WHERE id = %s', (post_id,))
    post = cursor.fetchone()
    
    if post is None:
        cursor.close()
        conn.close()
        flash('게시글을 찾을 수 없습니다.')
        return redirect(url_for('index'))
    
    cursor.execute('SELECT * FROM board.comments WHERE post_id = %s ORDER BY created_at', (post_id,))
    comments = cursor.fetchall()
    
    cursor.close()
    conn.close()
    
    user_ip = request.remote_addr
    conn = get_db_connection()
    cursor = conn.cursor()
    cursor.execute('SELECT COUNT(*) FROM board.likes WHERE post_id = %s AND user_ip = %s', (post_id, user_ip))
    liked = cursor.fetchone()[0] > 0
    cursor.close()
    conn.close()
    
    return render_template('view.html', post=post, comments=comments, liked=liked)

@app.route('/edit/<int:post_id>', methods=['GET'])
def edit_form(post_id):
    conn = get_db_connection()
    cursor = conn.cursor(cursor_factory=DictCursor)
    cursor.execute('SELECT * FROM board.posts WHERE id = %s', (post_id,))
    post = cursor.fetchone()
    cursor.close()
    conn.close()
    
    if post is None:
        flash('게시글을 찾을 수 없습니다.')
        return redirect(url_for('index'))
    
    return render_template('edit.html', post=post)

@app.route('/edit/<int:post_id>', methods=['POST'])
def edit_post(post_id):
    title = request.form.get('title')
    content = request.form.get('content')
    
    if not title or not content:
        flash('제목과 내용을 모두 입력해주세요.')
        return redirect(url_for('edit_form', post_id=post_id))
    
    conn = get_db_connection()
    cursor = conn.cursor()
    cursor.execute(
        'UPDATE board.posts SET title = %s, content = %s, updated_at = %s WHERE id = %s',
        (title, content, datetime.now(), post_id)
    )
    cursor.close()
    conn.close()
    
    flash('게시글이 성공적으로 수정되었습니다.')
    return redirect(url_for('view_post', post_id=post_id))

@app.route('/delete/<int:post_id>', methods=['POST'])
def delete_post(post_id):
    conn = get_db_connection()
    cursor = conn.cursor()
    cursor.execute('DELETE FROM board.posts WHERE id = %s', (post_id,))
    cursor.close()
    conn.close()
    
    flash('게시글이 성공적으로 삭제되었습니다.')
    return redirect(url_for('index'))

@app.route('/post/comment/<int:post_id>', methods=['POST'])
def add_comment(post_id):
    author = request.form.get('author')
    content = request.form.get('content')
    
    if not author or not content:
        flash('작성자와 내용을 모두 입력해주세요.')
        return redirect(url_for('view_post', post_id=post_id))
    
    conn = get_db_connection()
    cursor = conn.cursor()
    cursor.execute(
        'INSERT INTO board.comments (post_id, author, content) VALUES (%s, %s, %s)',
        (post_id, author, content)
    )
    cursor.close()
    conn.close()
    
    flash('댓글이 등록되었습니다.')
    return redirect(url_for('view_post', post_id=post_id))

@app.route('/post/like/<int:post_id>', methods=['POST'])
def like_post(post_id):
    user_ip = request.remote_addr
    
    conn = get_db_connection()
    cursor = conn.cursor()
    cursor.execute('SELECT COUNT(*) FROM board.likes WHERE post_id = %s AND user_ip = %s', (post_id, user_ip))
    already_liked = cursor.fetchone()[0] > 0
    
    if already_liked:
        cursor.execute('DELETE FROM board.likes WHERE post_id = %s AND user_ip = %s', (post_id, user_ip))
        cursor.execute('UPDATE board.posts SET like_count = like_count - 1 WHERE id = %s', (post_id,))
        message = '좋아요가 취소되었습니다.'
    else:
        cursor.execute('INSERT INTO board.likes (post_id, user_ip) VALUES (%s, %s)', (post_id, user_ip))
        cursor.execute('UPDATE board.posts SET like_count = like_count + 1 WHERE id = %s', (post_id,))
        message = '좋아요가 등록되었습니다.'
    
    cursor.close()
    conn.close()   
    flash(message)
    return redirect(url_for('view_post', post_id=post_id))

if __name__ == '__main__':
    app.run(debug=True)

데이터베이스 생성

-- board 스키마 생성
CREATE SCHEMA IF NOT EXISTS board;
-- 게시글 테이블
CREATE TABLE IF NOT EXISTS board.posts (
    id SERIAL PRIMARY KEY,
    title VARCHAR(255) NOT NULL,
    content TEXT NOT NULL,
    author VARCHAR(100) NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    view_count INTEGER DEFAULT 0,
    like_count INTEGER DEFAULT 0
);
-- 댓글 테이블
CREATE TABLE IF NOT EXISTS board.comments (
    id SERIAL PRIMARY KEY,
    post_id INTEGER REFERENCES board.posts(id) ON DELETE CASCADE,
    author VARCHAR(100) NOT NULL,
    content TEXT NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
-- 좋아요 테이블 (중복 좋아요 방지)
CREATE TABLE IF NOT EXISTS board.likes (
    id SERIAL PRIMARY KEY,
    post_id INTEGER REFERENCES board.posts(id) ON DELETE CASCADE,
    user_ip VARCHAR(45) NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    UNIQUE(post_id, user_ip)
);

참고로, SSL 연결은 Azure 연결을 위해서는 required로 해야하지만, local에서는 disabled되어있어야 연결 오류가 나지 않는다.
→required 한 경우 SSL 에러

플라스크 디버깅 모드 켜기

set FLASK_DEBUG=true

• 코드가 바뀔때마다 플라스크를 재실행하지 않아도 됨

농장관리 시스템을 위한 기능 추가

@app.route('/fms/result')
def fms_result():
    """total_result 뷰 데이터를 fms_result.html에 전달"""
    conn = get_db_connection()
    cursor = conn.cursor(cursor_factory=DictCursor)
    cursor.execute("SELECT * FROM fms.total_result ORDER BY 도착일 DESC NULLS LAST")
    results = cursor.fetchall()
    cursor.close()
    conn.close()
    return render_template('fms_result.html', results=results)

Azure로 데이터 옮기기

https://portal.azure.com/#home

Azure Database for PostgreSQL 유연한 서버

컴퓨팅+스토리지 서버 구성

네트워크

보안

만들기

DBeaver에서 PostgreSQL 백업

Azure Postgresql로 이전하기

DBeaver에서 연결

• azure에서 생성시 입력했던 유저이름과 비밀번호 입력

DBeaver에서 Restore

env파일 수정

DB_NAME='postgres'
DB_USER='azureuser'
DB_HOST='3dt016-boarddb.postgres.database.azure.com'
DB_PORT=5432
DB_PASSWORD='비번'

시각화 및 꾸미기

https://github.com/RudinP/MicrosoftDataSchool-myboardPractice