빠르게 활용하는 파이썬 3.2 프로그래밍 – 14장 데이터베이스

Notice : 해당 자료가 저작권등에 의해서 문제가 있다면 바로 삭제하겠습니다.

▣ 14장 데이터베이스

SQLite3 은 디스크 기반의 가벼운 데이터베이스 라이브러리
트렌젝션 지원, 성능과 안정성 검증(안드로이드, 아이폰 등 널리사용)
pysqlite 모듈, 기분적으로 포함

모듈함수
– sqlite3.connect(database[, timeout, detect_types, isolation_level, check_same_thread, – factory, cached_statements]) : SQLite3 DB와 연결된 Connection 객체반환
– sqlite3.complete_statement(sql) : 세미콜론으로 끝나는 sql 문에 대해 True를 반환(구문확인은 안함)
– sqlite3.register_converter(typename, callable) : SQLite3에 저장된 자료를 사용자 정의 자료형으로 변환하는 함수를 등록

Connection 클래스 : 연결된 DB응 동작
– cursor([cursorClass]) : cursor 객체를 생성
– commit() : 트렌젝션의 변경내용을 DB에 반영
– rollback() : 트렌젝션의 변경내용을 DB에 반영하지 않음
– close() : DB 연결을 종료
– isolation_level : 트렌젝션 경리수준을 확인, 설정
– execute(sql[, parameters]) : 임시 cursor 객체에 sql 문을 실행
– executemany(sql[, parameters]) : 임시 cursor 객체에 동일한 sql 문에 파라미터를 변경하여 실행
– executescript(sql_script) : 임시 cursor 객체에 세미콜론으로 구분된 여러줄의 sql 문을 실행
– create_aggregate(name, num_params, aggregate_class) : 사용자 정의 집계함수를 생성
– create_collation(name, callable) : 문자열 정렬시 sql 에서 사용하될 이름(name)과 정렬함수를 지정
– iterdump() : 연결된 DB의 내용을 sQL 질의 현태로 출력할수 있는 이터레이터를 반환

Cursor 클래스 : 실제적으로 DB의 sql 구문을 실행시키고 조회된 결과를 가져옴
– execute(sql[, parameters]) : cursor 객체에 sql 문을 실행
– executemany(sql[, parameters]) : cursor 객체에 동일한 sql 문에 파라미터를 변경하여 실행
– executescript(sql_script) : cursor 객체에 세미콜론으로 구분된 여러줄의 sql 문을 실행
– fetchone() : 조회된 결과로 부터 데이터 1개를 반환
– fetchmany(size=cursor.arraysize) : 조회된 결과로 부터 입력받은 size 만큼의 데이터 리스트를 반환
– fetchall() : 조회된 결과 모두를 리스트형태로 반환

Row 클래스 : 조회된 결과 집합에서 튜플을 나타냄

14.1 데이터베이스 연결

import sqlite3
con = sqlite3.connect("test.db")

1 2	import sqlite3 con = sqlite3.connect("test.db")

데이터베이스 객체를 생성, DB 가 없으면, 새로 생성, 있으면 오픈

14.2 SQL문 수행

import sqlite3
con = sqlite3.connect("test.db")
cur = con.cursor()
cur.execute("insert into test value(?,?);", (name, number))

import sqlite3

con = sqlite3.connect("test.db")

cur = con.cursor()

cur.execute("insert into test value(?,?);", (name, number))

? 를 사용하는 경우 전달순서를 기억해 거기에 맞춰 시퀀스 객체를 전달

cur.execute("insert into test value(:test_name, :test_num);", ("test_name":name, "test_num":number))

1	cur.execute("insert into test value(:test_name, :test_num);", ("test_name":name, "test_num":number))

사전형식으로 전달

datalist = (('apollo89', '123'), ('test', '111'))
cur.executemany("insert into test value(?, ?);", datalist)

1 2	datalist = (('apollo89', '123'), ('test', '111')) cur.executemany("insert into test value(?, ?);", datalist)

리스트 객체를 이용해 레코드를 연속으로 입력

14.3 레코드 조회

cur.execute("select * from test;")
for row in cur :
	print(row)
cur.fetchone()
cur.fetchmany(2)

cur.execute("select * from test;")

for row in cur :

print(row)

cur.fetchone()

cur.fetchmany(2)

14.4 트랜잭션 처리
con.commit() 입력
자동커밋 모드 : con.isolatoin_level = None

14.5 레코드 정렬과 사용자 정렬 함수

def userOrdef(s1,s2) : 
	pass

create_collation('myorder', userOrdef)
cur.execute("select * from test order by name collate myorder")

def userOrdef(s1,s2) :

pass

create_collation('myorder', userOrdef)

cur.execute("select * from test order by name collate myorder")

14.6 SQLite3 내장 집계 함수
sqlite 에서 사용할수 있는 내장 집계함수
abs(x), length(x), lower(x), upper(x), min(x), max(x), ramdom(*), count(x), count(*), sum(x)

14.7 사용자정의 집계 함수
사용자정의 집계함수를 만드고자 할때 : create_aggregate() 사용해서 클래스를 등록
클래스는 반드시 step() 과 finalize() 메소드가 정의되어있어야 함
– step() : 메소드 등록시 지정된 개수인 인자를 전달 받음
– finalize() : 집계된 결과를 반환

14.8 자료형
SQLite3 : Python
NULL : None
INTEGER : int
REAL : float
TEXT : str, bytes
BLOB : buffer

14.9 사용자정의 자료형
– sqlite3.register_adapter(파이썬 자료형, 변환함수)
– sqlite3.register_converter(sqllite3 자료형, 변환함수)

# -*- coding: cp949 -*-
class Point(object):   
    def __init__(self, x, y):
        self.x, self.y = x, y

    def __repr__(self):             # Point 객체의 내용 출력     
        return "Point(%f, %f)" % (self.x, self.y)

import sqlite3
def PointAdapter(point):                # 클래스 객체에서 SQLite3 입력 가능한 자료형으로 변환      
    return "%f:%f" % (point.x, point.y)

def PointConverter(s):                  # SQLite3에서 조회한 결과를 클래스 객체로 변환 
    x, y = list(map(float, s.decode().split(":")))
    return Point(x, y)

sqlite3.register_adapter(Point, PointAdapter)          # 클래스 이름과 변환 함수 등록
sqlite3.register_converter("point", PointConverter)    # SQL 구문에서 사용할 자료형 이름과 변환함수 등록

p = Point(4, -3.2)           # 입력할 데이터(파이썬 클래스 객체)
p2 = Point(-1.4, 6.2)

# 암묵적으로 선언된 자료형으로 조회하도록 설정  
con = sqlite3.connect(":memory:", detect_types=sqlite3.PARSE_DECLTYPES)
cur = con.cursor()
cur.execute("create table test(p point)")             # point 자료형을 이용하여 테이블 생성

cur.execute("insert into test values (?)", (p, ))     # point 레코드 입력
cur.execute("insert into test(p) values (?)", (p2,))

cur.execute("select p from test")    # 테이블 조회 
print([r[0] for r in cur])           
cur.close()
con.close()

# -*- coding: cp949 -*-

class Point(object):

def __init__(self, x, y):

self.x, self.y = x, y

def __repr__(self): # Point 객체의 내용 출력

return "Point(%f, %f)" % (self.x, self.y)

import sqlite3

def PointAdapter(point): # 클래스 객체에서 SQLite3 입력 가능한 자료형으로 변환

return "%f:%f" % (point.x, point.y)

def PointConverter(s): # SQLite3에서 조회한 결과를 클래스 객체로 변환

x, y = list(map(float, s.decode().split(":")))

return Point(x, y)

sqlite3.register_adapter(Point, PointAdapter) # 클래스 이름과 변환 함수 등록

sqlite3.register_converter("point", PointConverter) # SQL 구문에서 사용할 자료형 이름과 변환함수 등록

p = Point(4, -3.2) # 입력할 데이터(파이썬 클래스 객체)

p2 = Point(-1.4, 6.2)

# 암묵적으로 선언된 자료형으로 조회하도록 설정

con = sqlite3.connect(":memory:", detect_types=sqlite3.PARSE_DECLTYPES)

cur = con.cursor()

cur.execute("create table test(p point)") # point 자료형을 이용하여 테이블 생성

cur.execute("insert into test values (?)", (p, )) # point 레코드 입력

cur.execute("insert into test(p) values (?)", (p2,))

cur.execute("select p from test") # 테이블 조회

print([r[0] for r in cur])

cur.close()

con.close()

14.10 데이터베이스 덤프 만들기
DB의 현재 상태를 백업(sql)

import sqlite3
con = sqlite3.connect(":memory:")
cur = con.cursor()

cur.execute("CREATE TABLE PhoneBook(Name text, PhoneNum text);")
cur.execute("INSERT INTO PhoneBook VALUES('Derick', '010-1234-5678');")
list = (('Tom', '010-543-5432'), ('DSP', '010-123-1234'))
cur.executemany("INSERT INTO PhoneBook VALUES(?, ?);", list)

for l in con.iterdump():
    print(l)

import sqlite3

con = sqlite3.connect(":memory:")

cur = con.cursor()

cur.execute("CREATE TABLE PhoneBook(Name text, PhoneNum text);")

cur.execute("INSERT INTO PhoneBook VALUES('Derick', '010-1234-5678');")

list = (('Tom', '010-543-5432'), ('DSP', '010-123-1234'))

cur.executemany("INSERT INTO PhoneBook VALUES(?, ?);", list)

for l in con.iterdump():

print(l)

14.11 명령어 프롬프트에서 SQLite3 관리하기

# -*- coding: cp949 -*-
import sqlite3
import sys
import re

# 데이터베이스 경로 설정 
if len(sys.argv) == 2:
	path = sys.argv[1]
else:
	path = ":memory:"

con = sqlite3.connect(path)
con.isolation_level = None	# 트랜잭션없이 자동 커밋이 되도록 설정
cur = con.cursor()

buffer = ""			# 쿼리 버퍼 

def PrintIntro():
	"프로그램 인트로 메세지"
	print("pysqlite의 command 프로그램입니다.")
	print("특수 명령어를 알고 싶으시면 '.help;'를 입력하세요.")
	print("SQL 구문은 ';'으로 끝나야 합니다.")

def PrintHelp():
	"도움말"
	print(".dump\t\t데이터베이스의 내용을 덤프합니다.")

def SQLDump(con, file=None):
	"데이터베이스 내용 덤프"
	if file != None:
		f = open(file, "w")
	else:
		f = sys.stdout

	for l in con.iterdump():
		f.write("{0}\n".format(l))

	if f != sys.stdout:
		f.close()

PrintIntro()		# 인트로 메세지 출력

while True:
	line = input("pysqlite>> ")		# 명령어 입력
	if buffer == "" and line == "":
		break;
	buffer += line
	
	if sqlite3.complete_statement(buffer):		# ';'으로 구문이 끝나는지 검사
		buffer = buffer.strip()

		if buffer[0]==".":		# 특수 명령어인 경우
			cmd = re.sub('[ ;]', ' ', buffer).split()
			if cmd[0] == '.help':
				PrintHelp()
			elif cmd[0] == '.dump':
				if len(cmd) == 2:
					SQLDump(con, cmd[1])
				else:
					SQLDump(con)
		else:					# 일반 SQL 구문인 경우
			try:
				buffer = buffer.strip()
				cur.execute(buffer)

				if buffer.lstrip().upper().startswith("SELECT"):	# SELECT 질의인 경우
					print(cur.fetchall())
			except sqlite3.Error as e:
				print("Error: ", e.args[0])
			else:
				print("구문이 성공적으로 수행되었습니다.") 
		buffer=""		# 입력 버퍼 초기화
con.close()
print("프로그램을 종료합니다. 야옹~")

# -*- coding: cp949 -*-

import sqlite3

import sys

import re

# 데이터베이스 경로 설정

if len(sys.argv) == 2:

path = sys.argv[1]

else:

path = ":memory:"

con = sqlite3.connect(path)

con.isolation_level = None # 트랜잭션없이 자동 커밋이 되도록 설정

cur = con.cursor()

buffer = "" # 쿼리 버퍼

def PrintIntro():

"프로그램 인트로 메세지"

print("pysqlite의 command 프로그램입니다.")

print("특수 명령어를 알고 싶으시면 '.help;'를 입력하세요.")

print("SQL 구문은 ';'으로 끝나야 합니다.")

def PrintHelp():

"도움말"

print(".dump\t\t데이터베이스의 내용을 덤프합니다.")

def SQLDump(con, file=None):

"데이터베이스 내용 덤프"

if file != None:

f = open(file, "w")

else:

f = sys.stdout

for l in con.iterdump():

f.write("{0}\n".format(l))

if f != sys.stdout:

f.close()

PrintIntro() # 인트로 메세지 출력

while True:

line = input("pysqlite>> ") # 명령어 입력

if buffer == "" and line == "":

break;

buffer += line

if sqlite3.complete_statement(buffer): # ';'으로 구문이 끝나는지 검사

buffer = buffer.strip()

if buffer[0]==".": # 특수 명령어인 경우

cmd = re.sub('[ ;]', ' ', buffer).split()

if cmd[0] == '.help':

PrintHelp()

elif cmd[0] == '.dump':

if len(cmd) == 2:

SQLDump(con, cmd[1])

else:

SQLDump(con)

else: # 일반 SQL 구문인 경우

try:

buffer = buffer.strip()

cur.execute(buffer)

if buffer.lstrip().upper().startswith("SELECT"): # SELECT 질의인 경우

print(cur.fetchall())

except sqlite3.Error as e:

print("Error: ", e.args[0])

else:

print("구문이 성공적으로 수행되었습니다.")

buffer="" # 입력 버퍼 초기화

con.close()

print("프로그램을 종료합니다. 야옹~")

Apollo89.com

아폴로씨의 잡다한 경험들..

빠르게 활용하는 파이썬 3.2 프로그래밍 – 14장 데이터베이스

댓글 남기기응답 취소

카테고리

최신 댓글

메타