[Python] Postgresql 연동 / 연결 코드 샘플

환경

Python 3.9.7
- modules : psycopg2

Postgresql (Container, version:alpine3.18)
- host : 127.0.0.1
- port : 5432
- database : mydb
- 컨테이너 배포 샘플코드

---

Connection Class

import psycopg2
import logging

class DBBaseController:

    def __init__(self, host:str="127.0.0.1", port:int=5432) -> None:
        self.__connect(host=host, port=port)

    def __connect(self, host:str="127.0.0.1", port:int=5432):

        try:
            self._connection = psycopg2.connect(
                host="127.0.0.1" if not host else host,        # localhost, 127.0.0.1 등 접속 호스트 주소
                port="5432" if not port else port,            # DB 접속 port
                dbname="mydb",                                # database 이름
                user="user_id",                                # DB 사용자 계정
                password="my_password"                        # DB 사용자 계정 비밀번호
            )

            self._cur = self._connection.cursor()

        except psycopg2.Error as e:
            logging.error("[ERROR] DB Connecting error occured - {}".format(e))

---

사용 예시 코드

BaseController 클래스를 상속받아 Dataset 테이블을 관리하는 Controller 정의 후 사용

1. Dataset Table 생성 (SQL)

SQL 명령 코드는 링크 참조

CREATE TABLE "public"."dataset" (
    "id" text DEFAULT 'nextval(''dataset_id_seq'')' NOT NULL,
    "name" text NOT NULL,
    "description" text,
    "time" timestamp NOT NULL,
    CONSTRAINT "dataset_pkey" PRIMARY KEY ("id")
) WITH (oids = false);

2. 상속 Class & Method 정의

SELECT / INSERT / UPDATE / DELETE

import DBBaseController

class DatasetController(DBBaseController):

    def __init__(self, host:str="127.0.0.1", port:int=5432) -> None:
        super().__init__(host=host, port=port)


    # SELECT
    def getDatasetById(self, id:str) -> dict:
        try:

            self._cur.execute(f"SELECT id, name, description, time \
                                FROM dataset \
                                WHERE id='{id}'")

            dataset = self._cur.fetchone()

            result = {
                'id' : dataset[0],
                'name' : dataset[1],
                'description' : dataset[2] if dataset[2] else "",
                'time' : dataset[3]
            } if dataset else None

            return result

        except psycopg2.Error as err:
            logging.error("[ERROR] Selection error :: {}".format(err))

        return None


    # INSERT
    def insert(self, id:str, name:str, description:str="") -> bool:

        try:
            self._cur.execute("""INSERT INTO dataset (id, name, description, time) \
                                VALUES (%s, %s, %s, now()) ON CONFLICT (id) DO NOTHING""", (id, name, description))
            self._connection.commit()

            logging.info("[SUCCESS] Dataset insertion - {}".format(id))

        except psycopg2.Error as err:
            logging.error("[ERROR] Dataset insertion error :: {}".format(err))
            return False

        return True


    # UPDATE
    def update(self, id:str, name:str, description:str=""):
        try:

            _name = f"name = '{name}'," if name else ""
            _description = f"description = '{description}'," if description else ""

            self._cur.execute("UPDATE dataset \
                                SET {} {} time = now() \
                                WHERE id='{}'".format(_name, _description, id))
            self._connection.commit()

        except psycopg2.Error as e:
            logging.error("[ERROR] DB :: Test :: updateTestinfo - {}".format(e))
            return False

        return True


    # DELETE
    def delete(self, id:str):
        try:
            self._cur.execute(f"DELETE FROM dataset \
                                WHERE id='{id}'")
            self._connection.commit()

            logging.info("[SUCCESS] Dataset deleted".format(id))

        except psycopg2.Error as e:
            logging.error("[ERROR] Deletion error - {}".format(e))
            return False

        return True

3. Test

if __name__ == "__main__":
    dc = DatasetController(host="127.0.0.1", port=5432)

    # INSERT
    result = dc.insert(id="1", name="data01", description="my_data")
    print(f"INSERT : {result}")

    # SELECT
    result = dc.getDatasetById(id="1")
    print(f"SELECT : {result}")

    # DELETE
    result = dc.delete(id="1")
    print(f"DELETE : {result}")

    # INSERT
    result = dc.insert(id="1", name="data01", description="my_data")
    print(f"INSERT : {result}")

    # UPDATE
    result = dc.update(id="1", name="data_updated", description="updated")
    print(f"UPDATE : {result}")

    # SELECT
    result = dc.getDatasetById(id="1")
    print(f"SELECT : {result}")

    print("done")