Python Postgresql 연동하기

개발 환경 :

Python: Anconda 4.8.3

Database: POSTGRES 13.1

pip version: psycopg2     2.8.6

에디터: Visual Code

 

 

오늘 할것은 Python에서 Postgresql  연동을 하고 간단하게 CRUD API를 구현할 예정입니다.

먼저 POSTGRES에 파이썬 라이브러리인 psycopg2를 인스톨해줍니다. 

설치한 라이브러리를 임폴트 해주겠습니다.

import psycopg2

 

psycopg2.connect 함수를 이용하여서 데이터를 연결해주겠습니다.

db = psycopg2.connect(host='localhost', dbname='testdb',user='postgres',password='password',port=5432)

해당 변수에 알맞은 설정값을 넣어주시면 됩니다. 

명령을 처리할때 사용하는 cursor함수도 구현해주겟습니다.

cursor=db.cursor()

SQL 명령을 처리하기위해 execute 함수를 구현해주겠습니다.

def execute(self,query,args={}):
    self.cursor.execute(query,args)
    row = self.cursor.fetchall()
    return row

데이터베이스에 트랜잭션 변화가 있을때 커밋을해주는 함수를 구현해주곘습니다.

    def commit(self):
        self.cursor.commit()

 

저는 이것들을 하나의 클래스로써 생성하겠습니다.

import psycopg2

class Databases():
    def __init__(self):
        self.db = psycopg2.connect(host='localhost', dbname='test',user='postgres',password='password',port=5432)
        self.cursor = self.db.cursor()

    def __del__(self):
        self.db.close()
        self.cursor.close()

    def execute(self,query,args={}):
        self.cursor.execute(query,args)
        row = self.cursor.fetchall()
        return row

    def commit(self):
        self.cursor.commit()

다음은 CRUD(Create Read Update Delete) 문을 구현하겠습니다.

 

Create(Insert 문입니다)

def insertDB(self,schema,table,colum,data):
    sql = " INSERT INTO {schema}.{table}({colum}) VALUES ('{data}') ;".format(schema=schema,table=table,colum=colum,data=data)
    try:
        self.cursor.execute(sql)
        self.db.commit()
    except Exception as e :
        print(" insert DB  ",e) 

데이터베이스 테이블에 알맞은 insert 하는 로직입니다 

인자값으로는 스키마 , 테이블 ,컬럼 , 넣고자하는 데이터 가 있습니다.

Read(select 문입니다)

def readDB(self,schema,table,colum):
    sql = " SELECT {colum} from {schema}.{table}".format(colum=colum,schema=schema,table=table)
    try:
        self.cursor.execute(sql)
        result = self.cursor.fetchall()
    except Exception as e :
        result = (" read DB err",e)
    
    return result

ㄹ데이터베이스 테이블에서 원하고자 하는 값을 read하는 로직입니다.

인자값으로는 스키마, 테이블 , 조회하고자하는 컬럼  이 있습니다.

Update(update 문입니다)

def updateDB(self,schema,table,colum,value,condition):
    sql = " UPDATE {schema}.{table} SET {colum}='{value}' WHERE {colum}='{condition}' ".format(schema=schema
    , table=table , colum=colum ,value=value,condition=condition )
    try :
        self.cursor.execute(sql)
        self.db.commit()
    except Exception as e :
        print(" update DB err",e)

데이터베이스에 테이블에서 원하고자 하는 값을 update(변경)하는 로직입니다

인자값으로는 스키마,테이블,컬럼, 변경하고 자 하는 값 , 조건 이 있습니다.

Delete(delete 문입니다)

def deleteDB(self,schema,table,condition):
    sql = " delete from {schema}.{table} where {condition} ; ".format(schema=schema,table=table,
    condition=condition)
    try :
        self.cursor.execute(sql)
        self.db.commit()
    except Exception as e:
        print( "delete DB err", e)

데이터베이스에서 원하고자 하는 값을 delete(제거)하는 로직입니다

인자값으로는 스키마,테이블,조건 이 있습니다.

 

저는 이것들을 하나의 클래스로 집합해서 구현하면

from database import Databases

class CRUD(Databases):
    def insertDB(self,schema,table,colum,data):
        sql = " INSERT INTO {schema}.{table}({colum}) VALUES ('{data}') ;".format(schema=schema,table=table,colum=colum,data=data)
        try:
            self.cursor.execute(sql)
            self.db.commit()
        except Exception as e :
            print(" insert DB err ",e) 
    
    def readDB(self,schema,table,colum):
        sql = " SELECT {colum} from {schema}.{table}".format(colum=colum,schema=schema,table=table)
        try:
            self.cursor.execute(sql)
            result = self.cursor.fetchall()
        except Exception as e :
            result = (" read DB err",e)
        
        return result

    def updateDB(self,schema,table,colum,value,condition):
        sql = " UPDATE {schema}.{table} SET {colum}='{value}' WHERE {colum}='{condition}' ".format(schema=schema
        , table=table , colum=colum ,value=value,condition=condition )
        try :
            self.cursor.execute(sql)
            self.db.commit()
        except Exception as e :
            print(" update DB err",e)

    def deleteDB(self,schema,table,condition):
        sql = " delete from {schema}.{table} where {condition} ; ".format(schema=schema,table=table,
        condition=condition)
        try :
            self.cursor.execute(sql)
            self.db.commit()
        except Exception as e:
            print( "delete DB err", e)
if __name__ == "__main__":
    db = CRUD()
    db.insertDB(schema='myschema',table='table',colum='ID',data='유동적변경')
    print(db.readDB(schema='myschema',table='table',colum='ID'))
    db.updateDB(schema='myschema',table='table',colum='ID', value='와우',condition='유동적변경')
    db.deleteDB(schema='myschema',table='table',condition ="id != 'd'")

 이런 식 으로 사용할 수 있습니다.

실행 결과입니다. 

 

Insert 후 read 

그 후 update 후 delete 한 것을 데이터베이스에서 직접 출력해보았습니다.

 

잘못된 점이나 부족한 점 있으면 댓글 달아주시면 감사하겠습니다.