Review dit

database.py

@@ -0,0 +1,35 @@
+from psycopg2.pool import ThreadedConnectionPool
+from functools import lru_cache
+
+
+class Database:
+    def __init__(self, database: str = 'postgres', host: str = 'localhost', password: str = 'secret', user: str = 'root', port: str = '5432'):
+        self.__database = database
+        self.__host = host
+        self.__password = password
+        self.__user = user
+        self.__port = port
+        self.__pool = self.__create_connection()
+
+    def __create_connection(self):
+        try:
+            pool_connection = ThreadedConnectionPool(
+                minconn=1,
+                maxconn=10,
+                dsn=f'postgresql://{self.__user}:{self.__password}@{self.__host}:{self.__port}/{self.__database}?application_name=python-postgresql-conn'
+            )
+            return pool_connection
+        except Exception as e:
+            raise ValueError(e)
+
+    def get_pool_conn(self):
+        return self.__pool.getconn()
+
+    def put_pool_conn(self, conn):
+        return self.__pool.putconn(conn)
+
+database_obj = Database()
+
+@lru_cache
+def get_connection() -> Database:
+    return database_obj

main.py

@@ -0,0 +1,44 @@
+from database import get_connection
+
+
+class RepositoryPostgresql:
+    def __init__(self):
+        self.__conn = get_connection()
+
+    def get_all_data(self):
+        pool = self.__conn.get_pool_conn()
+        try:
+            cur = pool.cursor()
+            sql = """
+                select * from tbl_job_desc limit 10
+            """
+            cur.execute(sql)
+            return cur.fetchall()
+        except Exception as error:
+            raise error
+        finally:
+            self.__conn.put_pool_conn(pool)
+
+    def get_one_data(self, id):
+        pool = self.__conn.get_pool_conn()
+        try:
+            cur = pool.cursor()
+            sql = """
+                select * from tbl_job_desc where id = %s limit 1
+            """%(id)
+            cur.execute(sql)
+            return cur.fetchone()
+        except Exception as error:
+            raise error
+        finally:
+            self.__conn.put_pool_conn(pool)
+
+
+if __name__ == '__main__':
+    databaseRepo = RepositoryPostgresql()
+
+    data = databaseRepo.get_all_data()
+    print(data)
+
+    data = databaseRepo.get_one_data(1)
+    print(data)