SQLite 링크: https://www.sqlite.org/index.html
Python SQLite : https://docs.python.org/ko/3/library/sqlite3.html
- 내장 가능한 오픈소스 데이터베이스이다. C로 작성됐으며 일반적인 SQL로 쿼리가 가능한다.
- SQL은 다양한 플랫폼에서 이식됨.
1. 장점
- 관계형 데이터베이스 역할을 하고, 트랜잭션을 지원함.
- 전체 텍스트 인덱스, JSON 데이터 지원 등 고급 데이터베이스 기능을 지원함.
- 단일 독립형 바이너리이므로 앱과 함께 배포하고 앱과 이동하기 쉽다.
2. 특징
- 데이터형식은 TEXT, NUMERIC, INTEGER, REAL, BLOB 로 소수다. 일자는 함수를 제공한다.
https://www.sqlite.org/lang_datefunc.html
- SQLite는 단일 애플리케이션이 적합하고, 동시 사용자는 MySQL 또는 내장DB중에는 버클리DB가 적당함.
버클리DB 엔진 성능은 SQLite를 못 따라감.
9. 참조
9.1 isolation_level : https://www.sqlite.org/lang_transaction.html
- None : autommit
- 'DEFERRED': (default) 쓰기모드에서 트랜잭션 시작됨. commit / rollback 처리 필요.
- 'IMMEDIATE': 데이터베이스 연결 즉시 트랜잭션 시작.
- 'EXCLUSIVE': IMMEDIATE와 유사하지만 트랜잭션 동안 다른 데이터베이스 연결이 데이터베이스 읽는 것을 방지함.
9.2 다중 스레드 응용 프로그램에서 SQLite 사용
- 관련글: https://docs.python.org/ko/3/library/sqlite3.html (sqlite3.threadsafety)
https://www.sqlite.org/threadsafe.html
- python / sqlite3 에서 qlite3.threadsafety=1 로 스레드는 모듈을 공유할 수 있지만, 연결을 공유할 수 없음.
연결 시 check_same_thread=True(기본값)를 False로 변경하면, 다른 스레드로 공유가능함.
단, 쓰기에서는 직렬화 해서 사용해야 안전한데 현재 sqlite3에서는 지원이 안되는 것 같음.
import sqlite3
# 실행위치에 파일이 생성되고, 상대위치를 지정하면 실행 최상위를 계산해서 수행함.
conn = sqlite3.connect('./universe_price.db', isolation_level=None)
cur = conn.cursor()
def create_table():
cur.execute('''CREATE TABLE IF NOT EXISTS
BALANCE
(
CODE VARCHAR(6) PRIMARY KEY,
BID_PRICE INT(20) NOT NULL,
QUANTITY INT(20) NOT NULL,
CREATED_AT VHARCHAR(14) NOT NULL,
WILL_CLEAR_AT VARCHAR(14)
)
''')
# 단건 등록 테스트
def test_insert_row():
sql = """INSERT INTO BALANCE(CODE, BID_PRICE, QUANTITY, CREATED_AT, WILL_CLEAR_AT) VALUES
(?, ?, ?, ?, ?)
"""
cur.execute(sql, ('005930', 70000, 10, '20201222', 'today'))
# conn.commit()
print(cur.rowcount)
# 복수건 등록 테스트
def test_insert_rows():
sql = """INSERT INTO BALANCE(CODE, BID_PRICE, QUANTITY, CREATED_AT, WILL_CLEAR_AT) VALUES
(?, ?, ?, ?, ?)
"""
stocks = [
('010060', 109500, 10, '20201222', 'today'),
('005380', 168000, 10, '20201222', 'today'),
('336260', 39700, 10, '20201222', 'today')
]
cur.executemany(sql, stocks)
# conn.commit()
print('insert row count: {}'.format(cur.rowcount))
# 단건 조회 테스트
def test_select_one():
cur.execute('SELECT * FROM BALANCE')
row = cur.fetchone()
print(row)
# 전체 조회 테스트
def test_select_all():
cur.execute('SELECT * FROM BALANCE WHERE BID_PRICE > :BID_PRICE', {'BID_PRICE':50000})
rows = cur.fetchall()
print('# 전체 조회 테스트')
for row in rows:
code, bid_price, quantity, created_at, will_clear_at = row
print(row)
# 일부건 패치 테스트
def test_select_many(row_count):
cur.execute('SELECT * FROM BALANCE')
rows = cur.fetchmany(row_count)
print('# 일부건 조회 테스트')
for row in rows:
code, bid_price, quantity, created_at, will_clear_at = row
print(row)
def close_conn():
conn.close()
if __name__ == "__main__":
create_table()
# test_insert_row()
# test_insert_rows()
test_select_all()
test_select_many(2)
close_conn()
end.