MySQL 보안 연결 & MyDB 클래스 & OpenAPI 실습 정리

syeom·2026년 4월 20일

데이터분석가 부트캠프 멀티캠퍼스 부트캠프 부트캠프

멀티캠퍼스 데이터분석_11회차

목록 보기

9/46

멀티캠퍼스 데이터분석 4월 20일 수업 내용 — dotenv, pymysql 직접 연결, DB 연동 클래스 설계, OpenAPI, DART API

📌 목차

파이썬 기초 복습 — 시험 관련
classmethod 데코레이터
dotenv — 민감 정보 숨기기
pymysql 직접 연결 실습
MyDB 클래스 설계
MyDB 사용 예시 — CRUD 전체
OpenAPI — requests로 데이터 받아오기
OpenAPI 연습 — xml & json 동시 요청
DART API — 기업 코드 로드 함수
MyDB 모듈 코드 전문 (db.py)

1. 파이썬 기초 복습 — 시험 관련

가변 데이터 함수 기본값 주의

def func(x, acc=[]):
    acc.append(x)
    return acc

print(func(1))      # [1]
print(func(2))      # [1, 2]  ← acc가 유지됨!
print(func(3, []))  # [3]     ← 새 리스트 전달 시 독립적

⚠️ 가변 데이터를 기본값으로 사용하면 함수 호출 간에 상태가 유지됩니다.

얕은 복사 주의

a = [1, 2, 3]
b = a            # 얕은 복사 — 같은 주소
b[0] = 100
print(a)         # [100, 2, 3] — a도 변경됨!

슬라이싱 — 종료 인덱스 미포함

print([1, 2, 3, 4, 5][1:4])   # [2, 3, 4]  — 4번 인덱스(5)는 미포함

bool 강제 변환

bool("")      # False  — 빈 문자열
bool("hello") # True
bool(0)       # False
bool(-1)      # True   — 음수도 True
bool([])      # False  — 빈 리스트
bool([1, 2])  # True

유효한 변수명 규칙

공백 사용 불가
_ 을 제외한 특수문자 사용 불가
첫 글자는 문자 또는 _ 로 시작 (숫자 시작 불가)
대소문자 구분

2. classmethod 데코레이터

클래스 변수를 수정할 때 클래스명을 직접 쓰는 대신 @classmethod 를 사용하면 클래스 이름이 바뀌어도 자동으로 따라갑니다.

class Test2:
    text = 'Hello'

    def __init__(self, a):
        self.a = a

    @classmethod
    def func_1(cls, x):
        # cls → 클래스 이름 자동 대입
        cls.text = x

class_3 = Test2(10)
class_4 = Test2(20)

Test2.func_1('Hi')

print(class_3.text)   # Hi
print(class_4.text)   # Hi

	일반 메서드	classmethod
첫 번째 매개변수	`self` (인스턴스)	`cls` (클래스)
클래스 변수 접근	`클래스명.변수`	`cls.변수`
클래스명 변경 시	직접 수정 필요	자동 반영

3. dotenv — 민감 정보 숨기기

DB 접속 정보를 코드에 직접 작성하면 GitHub 업로드 시 보안 사고가 발생할 수 있습니다.

❌ 코드에 비밀번호 직접 입력 → GitHub 업로드 → 노출
✅ .env 파일에 저장 → .gitignore에 추가 → 안전하게 관리

설치 & 사용

# !pip install python-dotenv
from dotenv import load_dotenv
import os

load_dotenv()          # .env 파일의 내용을 환경 변수로 임시 등록

os.getenv('host')      # 값 꺼내기
os.getenv('port')      # 반환 타입은 항상 문자열 → 숫자 필요 시 int() 변환

.env 파일 예시

host=127.0.0.1
port=3306
user=root
pwd=1234
db_name=multicam

⚠️ .env 파일은 반드시 .gitignore에 추가해서 GitHub에 올라가지 않도록 해야 합니다.

4. pymysql 직접 연결 실습

MyDB 클래스 없이 pymysql로 직접 연결하고 쿼리를 실행하는 기본 흐름입니다.

연결 & Cursor 생성

import pymysql

_db = pymysql.connect(
    host     = os.getenv('host'),
    port     = int(os.getenv('port')),   # getenv() 결과는 문자열 → int 변환 필수
    user     = os.getenv('user'),
    password = os.getenv('pwd'),
    db       = os.getenv('db_name')
)

# 기본 Cursor — 결과가 튜플 형태
cursor = _db.cursor()

# DictCursor — 결과가 딕셔너리 형태 (컬럼명이 key) ✅ 권장
cursor = _db.cursor(pymysql.cursors.DictCursor)

Cursor 종류	결과 타입	DataFrame 변환 시 컬럼명
기본 `cursor()`	튜플	자동 생성 (0, 1, 2...)
`DictCursor`	딕셔너리	DB 컬럼명 자동 적용 ✅

CREATE TABLE

create_table = """
    CREATE TABLE IF NOT EXISTS `user_info` (
        `id`       VARCHAR(32) PRIMARY KEY,
        `password` VARCHAR(32) NOT NULL,
        `name`     VARCHAR(32),
        `age`      INT
    )
"""
cursor.execute(create_table)

💡 DDL(CREATE, ALTER, DROP)은 commit() 없이 바로 DB에 반영됩니다.

INSERT — 고정값

sign_up_query = """
    INSERT INTO `user_info`
    VALUES ('test', '1234', 'kim', 30)
"""
cursor.execute(sign_up_query)
_db.commit()   # DML은 commit() 필수

INSERT — %s 파라미터 방식 ✅ 권장

매번 쿼리를 새로 작성하는 번거로움을 없앱니다.

signup_query = """
    INSERT INTO `user_info`
    VALUES (%s, %s, %s, %s)
"""

input_id   = input('아이디를 입력하시오')
input_pass = input('비밀번호를 입력하시오')
input_name = input('이름을 입력하시오')
input_age  = input('나이를 입력하시오')

# %s 위치에 리스트의 원소들이 순서대로 대입
cursor.execute(signup_query, [input_id, input_pass, input_name, input_age])
_db.commit()

💡 %s 의 개수와 데이터 개수는 반드시 일치해야 합니다.

SELECT — fetchall()

user_list_query = "SELECT * FROM `user_info`"
cursor.execute(user_list_query)
result = cursor.fetchall()   # 결과를 가져온 뒤 cursor 내부 데이터는 제거됨

💡 SELECT vs CUD 차이

SELECT → fetchall()로 결과를 받아옴

INSERT / UPDATE / DELETE → 반환 데이터 없음, _db.commit()으로 DB에 반영

연결 종료

_db.close()

5. MyDB 클래스 설계

매번 연결·커서 생성·종료를 반복하는 번거로움을 없애기 위해 DB 연동 클래스를 설계합니다.

클래스 선언 조건

생성자 함수 (__init__)

서버의 정보를 입력받아 저장
클래스가 생성될 때마다 다른 DB 서버 정보를 담을 수 있음
2개의 객체를 생성해서 서로 다른 DB 서버에 동시 연동 가능

sql_query(query, *datas) 함수

서버와 연결 → Cursor 생성
CUD (INSERT, UPDATE, DELETE) → execute()로 질의 → Cursor의 데이터 변화
R (SELECT) → execute() 후 fetchall()로 결과 반환
쿼리 시작이 select라면 자동으로 fetchall() 호출 후 리턴

commit() 함수

_db.commit()으로 데이터 확정 (동기화)
_db.close()로 서버 연결 종료
del self.db_server로 변수 자체를 제거

설계 목표

함수	역할
`__init__()`	서버 정보 저장
`sql_query(query, *datas)`	쿼리 실행 (SELECT → 결과 반환 / CUD → 'Query Ok!')
`commit()`	DB 확정 + 연결 종료 + 변수 제거

핵심 설계 포인트

# 이미 접속 중이면 재접속하지 않는다
try:
    self.db_server   # 변수가 있으면 → 이미 접속 중
    print('접속된 서버가 존재함')
except:
    # 변수가 없으면 NameError → 새로 연결
    self.db_server = pymysql.connect(...)

# SELECT 쿼리 자동 감지
if query.lstrip().lower().startswith('select'):
    result = cursor.fetchall()
else:
    result = 'Query Ok!'

💡 self.변수 vs 지역 변수

self.변수 → 함수 종료 후에도 객체 안에 유지됨

일반 변수 → 함수 종료 시 사라짐 (휘발성)

클래스 전체 코드

class MyDB:
    # 생성자 함수 (서버의 정보를 받아오기 위해 사용)
    def __init__(self, host, port, user, password, db):
        self.host     = host
        self.port     = port
        self.user     = user
        self.password = password
        self.db       = db

    # 데이터베이스에 변화를 주는 함수
    def commit(self):
        try:
            # DB 확정 (동기화)
            self.db_server.commit()
            # 연결 종료
            self.db_server.close()
            # close()만으로는 변수가 사라지지 않으므로 변수 자체를 제거
            del self.db_server
        except:
            print('데이터베이스 서버와의 연결이 되어있지 않습니다. sql_query() 함수를 호출하여 서버와의 연결을 해주세요')

    def sql_query(self, query, *datas):
        # 이미 접속 중인 경우 재접속 금지 (commit 전 데이터 유실 방지)
        try:
            self.db_server
            print('접속된 서버가 존재함')
        except:
            # self.db_server 변수가 없을 때만 새로 연결
            self.db_server = pymysql.connect(
                host     = self.host,
                port     = self.port,
                user     = self.user,
                password = self.password,
                db       = self.db
            )

        # DictCursor — 결과를 딕셔너리 형태로 반환
        cursor = self.db_server.cursor(pymysql.cursors.DictCursor)

        try:
            cursor.execute(query, datas)

            # 쿼리가 SELECT문이라면 결과값을 반환
            if query.lstrip().lower().startswith('select'):
                result = cursor.fetchall()
            else:
                result = 'Query Ok!'

            return result

        except Exception as e:
            print('query문 execute 중 에러')
            print(e)


# 클래스 생성 (서버 정보 등록)
db1 = MyDB(
    host     = os.getenv('host'),
    port     = int(os.getenv('port')),
    user     = os.getenv('user'),
    password = os.getenv('pwd'),
    db       = os.getenv('db_name')
)

6. MyDB 사용 예시 — CRUD 전체

from dotenv import load_dotenv
import os

load_dotenv()

db1 = MyDB(
    host     = os.getenv('host'),
    port     = int(os.getenv('port')),
    user     = os.getenv('user'),
    password = os.getenv('pwd'),
    db       = os.getenv('db_name')
)

CREATE TABLE

create_table = """
    CREATE TABLE IF NOT EXISTS `user_info` (
        `id`       VARCHAR(32) PRIMARY KEY,
        `password` VARCHAR(32) NOT NULL,
        `name`     VARCHAR(32),
        `age`      INT
    )
"""
db1.sql_query(create_table)

INSERT

insert_query = """
    INSERT INTO `user_info`
    VALUES (%s, %s, %s, %s)
"""
data_list = ['test3', '0000', 'lee', 40]
db1.sql_query(insert_query, *data_list)   # 언패킹으로 전달

SELECT

select_query = "SELECT * FROM `user_info`"
db1.sql_query(select_query)

UPDATE

update_query = """
    UPDATE `user_info` SET `password` = %s
    WHERE `id` = %s
"""
db1.sql_query(update_query, '0123', 'test3')
db1.sql_query(select_query)   # 결과 확인
db1.commit()                  # 확정

DELETE

delete_query = """
    DELETE FROM `user_info`
    WHERE `id` = %s
"""
db1.sql_query(delete_query, 'test')
db1.commit()

다른 DB 서버에 연결

db2 = MyDB(
    host     = os.getenv('host2'),
    port     = int(os.getenv('port2')),
    user     = os.getenv('user2'),
    password = os.getenv('pwd2'),
    db       = os.getenv('db_name2')
)
db2.sql_query('SELECT * FROM `AAPL`')

외부 모듈로 분리 후 사용

import db
import importlib

importlib.reload(db)   # 수정된 모듈 재로드

db3 = db.MyDB()
db3.sql_query(select_query)

⚠️ commit() 전에 재접속 금지
INSERT 후 commit() 없이 sql_query()를 다시 호출하면 새로 연결되어 이전 데이터가 사라집니다.

⚠️ 트랜잭션 Lock 주의
commit() 또는 rollback() 없이 연결이 끊기면 Lock이 걸릴 수 있습니다.
PRIMARY KEY가 없는 테이블에서도 발생할 수 있으므로 주의하세요.

7. OpenAPI — requests로 데이터 받아오기

OpenAPI는 웹 주소(URL)로 데이터를 요청하면 서버가 데이터를 보내주는 방식입니다.

# !pip install requests
import requests
import json
from pprint import pprint

load_dotenv()
service_key = os.getenv('service_key')

url = 'https://apis.data.go.kr/B553069/esg/corp/getEsgStatus'
# 파라미터 지정 (dict형으로 key 값들은 openapi가 원하는 parameter의 key값들을 지정 )
params = {
    'serviceKey' : service_key,
    'pageNo'     : 1,
    'numOfRows'  : 100
}
# 서버에게 요청을 보낸다
# requests 안에 get() 함수를 이용
    # get()
        # 첫번째 인자 -> 주소
        # params 매개변수 -> 파라미터 값들을 dict 형태로 대입
res      = requests.get(url, params=params)
res_data = json.loads(res.content)   # bytes → dict 변환
pprint(res_data)

import pandas as pd
df = pd.DataFrame(res_data['response']['body']['items'])

💡 res.content → bytes 타입
json.loads() → dict 타입으로 변환 후 키로 데이터 접근

8. OpenAPI 연습 — xml & json 동시 요청

openapi 연습

한국자산관리공사_종전부동산 관련 현황 조회서비스 api를 이용해서 데이터를 불러오자
serviceKey는 dotenv에 저장하여 불러오는 형태로 사용
페이지는 1페이지, 페이지당 데이터수는 100
xml data와 json 데이터를 하나씩 저장(resultType 변경하여 2번 요청)
json 데이터를 이용하여 데이터프레임으로 결과를 만드시오.

url = 'http://apis.data.go.kr/B010003/kamcoRlctRlst/pscd'

# JSON 요청
params_json = {
    'serviceKey' : os.getenv('serviceKey'),
    'pageNo'     : 1,
    'numOfRows'  : 100,
    'resultType' : 'json'
}
res_json      = requests.get(url, params=params_json)
res_data_json = json.loads(res_json.content)
df_json = pd.DataFrame(res_data_json['body']['items']['item'])

# XML 요청
params_xml = {
    'serviceKey' : os.getenv('serviceKey'),
    'pageNo'     : 1,
    'numOfRows'  : 100,
    'resultType' : 'xml'
}
res_xml = requests.get(url, params=params_xml)

# !pip install xmltodict
from xmltodict import parse
res_data_xml = parse(res_xml.content)
df_xml = pd.DataFrame(res_data_xml['response']['body']['items']['item'])

응답 타입	변환 라이브러리	변환 함수
JSON	`json` (기본 제공)	`json.loads(res.content)`
XML	`xmltodict` (설치 필요)	`parse(res.content)`

9. DART API — 기업 코드 로드 함수

금융감독원 DART API에서 기업 코드를 ZIP 파일로 받아 파싱하는 함수입니다.

import os, io, zipfile, requests
import xml.etree.ElementTree as ET
from dotenv import load_dotenv

load_dotenv()
API_KEY = os.getenv('dart')

def get_corp_code_dict():
    url    = 'https://opendart.fss.or.kr/api/corpCode.xml'
    params = {'crtfc_key': API_KEY}

    res = requests.get(url, params=params)

    with zipfile.ZipFile(io.BytesIO(res.content)) as zf:
        xml_content = zf.read(zf.namelist()[0])

    root      = ET.ElementTree(ET.fromstring(xml_content)).getroot()
    corp_dict = {}

    for child in root:
        corp_dict[child.findtext('corp_name')] = child.findtext('corp_code')

    return corp_dict

corp_dict = get_corp_code_dict()

처리 흐름

API 요청
    ↓
ZIP 파일 (bytes)
    ↓
zipfile.ZipFile(io.BytesIO(...))  — 메모리에서 바로 압축 해제
    ↓
XML 파일 추출 → ElementTree 파싱
    ↓
{기업명: 기업코드} 딕셔너리 생성

10. MyDB 모듈 코드 전문 (db.py)

import pymysql

class MyDB:
    def __init__(self, host, port, user, password, db):
        self.host     = host
        self.port     = port
        self.user     = user
        self.password = password
        self.db       = db

    def commit(self):
        try:
            self.db_server.commit()
            self.db_server.close()
            del self.db_server
        except:
            print('DB 서버와 연결되어 있지 않습니다. sql_query()를 먼저 호출하세요.')

    def sql_query(self, query, *datas):
        # 이미 접속 중이면 재접속 금지
        try:
            self.db_server
            print('접속된 서버가 존재함')
        except:
            self.db_server = pymysql.connect(
                host     = self.host,
                port     = self.port,
                user     = self.user,
                password = self.password,
                db       = self.db
            )

        cursor = self.db_server.cursor(pymysql.cursors.DictCursor)

        try:
            cursor.execute(query, datas)

            if query.lstrip().lower().startswith('select'):
                result = cursor.fetchall()
            else:
                result = 'Query Ok!'

            return result

        except Exception as e:
            print('query 실행 중 에러:', e)

📎 핵심 개념 요약

개념	설명
`python-dotenv`	`.env` 파일로 민감 정보 관리
`load_dotenv()`	`.env` 파일 환경 변수 등록
`os.getenv('키')`	등록된 환경 변수 값 가져오기
`pymysql.connect(...)`	DB 서버 연결
`cursor()`	기본 커서 (튜플 반환)
`cursor(DictCursor)`	딕셔너리 형태로 결과 반환
`cursor.execute(q, data)`	쿼리 + `%s` 파라미터 실행
`cursor.fetchall()`	SELECT 결과 전체 가져오기
`_db.commit()`	INSERT/UPDATE/DELETE 확정
`_db.close()`	연결 종료
`requests.get(url, params)`	OpenAPI GET 요청
`json.loads(res.content)`	bytes → dict 변환
`xmltodict.parse()`	XML bytes → dict 변환
`@classmethod`	클래스 이름을 `cls`로 자동 전달
`zipfile.ZipFile(io.BytesIO(...))`	메모리에서 ZIP 파일 처리

syeom

공부 기록

이전 포스트

데이터베이스 & Python 연결 — 이론 & 실습 정리

다음 포스트

MySQL 보안 연결 & MyDB 클래스 & OpenAPI 실습 정리

멀티캠퍼스 데이터분석_11회차

📌 목차

1. 파이썬 기초 복습 — 시험 관련

가변 데이터 함수 기본값 주의

얕은 복사 주의

슬라이싱 — 종료 인덱스 미포함

bool 강제 변환

유효한 변수명 규칙

2. classmethod 데코레이터

3. dotenv — 민감 정보 숨기기

설치 & 사용

.env 파일 예시

4. pymysql 직접 연결 실습

연결 & Cursor 생성

CREATE TABLE

INSERT — 고정값

INSERT — %s 파라미터 방식 ✅ 권장

SELECT — fetchall()

연결 종료

5. MyDB 클래스 설계

클래스 선언 조건

설계 목표

핵심 설계 포인트

클래스 전체 코드

6. MyDB 사용 예시 — CRUD 전체

CREATE TABLE

INSERT

SELECT

UPDATE

DELETE

다른 DB 서버에 연결

외부 모듈로 분리 후 사용

7. OpenAPI — requests로 데이터 받아오기

8. OpenAPI 연습 — xml & json 동시 요청

openapi 연습

9. DART API — 기업 코드 로드 함수

처리 흐름

10. MyDB 모듈 코드 전문 (db.py)

📎 핵심 개념 요약

데이터베이스 & Python 연결 — 이론 & 실습 정리

플라스크 & HTML & 웹 크롤링 — 이론 & 실습 정리

0개의 댓글