☀️ haejun

Today-·Total-

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

3 min read

Summary

  • sqlalchemy의 create_engine으로 DB 엔진을 만들고, 비밀번호에 특수문자가 있으면 quote_plus로 인코딩한다
  • pd.read_sql로 조회하고 to_sql로 저장하며, 대용량은 LIMIT/OFFSET으로 나눠 불러온다
  • MySQL은 쿼리를 text()로 감싸야 한다

PostgreSQL 연결

기본 세팅

sqlalchemy와 psycopg2를 임포트한다. 비밀번호에 특수문자가 포함되면 quote_plus()로 인코딩한 뒤 엔진에 연결한다.

from sqlalchemy import create_engine, text
import pandas as pd
import psycopg2
from urllib.parse import quote_plus
 
user = 'postgres'
password = quote_plus('비밀번호@3$')   # 특수문자 인코딩
host = '10.10.12.181'
port = '5432'
dbname = 'dataops'
 
postgres_engine = create_engine(
    f'postgresql+psycopg2://{user}:{password}@{host}:{port}/{dbname}'
)

데이터 입출력

pd.read_sql로 쿼리 결과를 불러온다.

query = 'SELECT * FROM schemas.table_name'
db_data = pd.read_sql(query, postgres_engine)

to_sql로 저장한다. DB에 미리 생성된 테이블에, 열 이름도 동일하게 지정해야 한다.

  • name: 저장할 테이블 이름
  • con: DB 엔진 객체
  • schema: 저장할 스키마 이름
  • index: 데이터프레임 인덱스 저장 여부(False면 저장 안 함)
  • if_exists: 테이블이 있을 때 동작 — replace(삭제 후 새로 저장), append(밑에 추가), fail(오류 발생)
db_data.to_sql(name='table_name', con=postgres_engine,
               if_exists='append', index=False, schema='schema_name')

shp 파일 저장

GeoPandas의 to_postgis()를 활용한다(GeoAlchemy2 패키지 필요). 매개변수는 to_sql()과 동일하다.

대용량 데이터 나눠 불러오기

LIMIT(불러올 행 개수)과 OFFSET(건너뛸 행 개수)으로 전체를 일정 크기씩 나눠 불러와 결합한다.

from tqdm import tqdm
 
chunk_size = 10000
offset = 0
pbar = tqdm(total=None)
df = pd.DataFrame()
 
try:
    while True:
        query = f'SELECT * FROM schemas.table_name LIMIT {chunk_size} OFFSET {offset}'
        chunk_df = pd.read_sql(query, postgres_engine)
        offset += chunk_size
 
        if chunk_df.empty:
            break
 
        df = pd.concat([df, chunk_df], ignore_index=True)
        pbar.update(len(chunk_df))
except KeyboardInterrupt:
    print('Loop interrupted.')
finally:
    pbar.close()

스키마 내 테이블 목록 조회

information_schema.tables를 조회해 특정 스키마의 테이블 목록을 가져온다.

import re
 
def schema_tables(db_schema):
    cursor = conn.cursor()
    query = "SELECT table_name FROM information_schema.tables WHERE table_schema = %s;"
    cursor.execute(query, [db_schema])
    tables = cursor.fetchall()
    cursor.close()
    return [item[0] for item in tables]
 
db_table_list = schema_tables('full_text')
pattern = re.compile(r'.*shinhan.*info$')
shinhan_list = [item for item in db_table_list if pattern.match(item)]

MySQL 연결

text()로 쿼리 감싸기

MySQL은 기존 쿼리 입력 방식을 지원하지 않으므로, text()로 쿼리를 감싼다.

from sqlalchemy import create_engine
from sqlalchemy.sql import text
 
engine = create_engine('mysql+pymysql://user:password@host:3306/dbname?charset=utf8')
 
with engine.connect() as connection:
    connection.execute(text('CREATE TABLE tbl_upload LIKE tbl_source'))
    connection.execute(text('INSERT INTO tbl_upload SELECT * FROM tbl_source'))
    connection.execute(text('DROP TABLE IF EXISTS tbl_upload'))

Reference

Graph View