def start(self):
self.say()
self.eat()
self.walk()psycopg2.connect(...)
import psycopg2
from psycopg2 import pool
import datetime
class PotgressConn:
def __init__(self, config=None):
self.conf = config
pg_user = self.conf.get_config("Postgres.user")
pg_pass = self.conf.get_config("Postgres.password")
pg_host = self.conf.get_config("Postgres.host")
pg_port = self.conf.get_config("Postgres.port")
pg_db = self.conf.get_config("Postgres.database")
self.__pool = psycopg2.pool.SimpleConnectionPool(1, 20, user=pg_user,
password=pg_pass,
host=pg_host,
port=pg_port,
database=pg_db)
if self.__pool:
print("PostgreSQL: Connection pool created successfully")
def get_conn(self):
if self.__pool:
return self.__pool.getconn()
def disconnect(self):
if self.__pool:
self.__pool.closeall
print("PostgreSQL: connection pool is closed")
def execute(self, query, transaction_type = 'select', fetch_type='all'):
try:
records = None
ps_connection = self.__pool.getconn()
if ps_connection:
ps_cursor = ps_connection.cursor()
ps_cursor.execute(query)
if transaction_type in('insert', 'delete'):
ps_connection.commit()
else:
if fetch_type == "all":
records = ps_cursor.fetchall()
elif fetch_type == "many":
records = ps_cursor.fetchmany()
else:
print(ps_cursor.statusmessage)
records = ps_cursor.fetchone()
ps_cursor.close()
return records
except Exception as err:
raise err
finally:
self.disconnect()
def insert_one(self, table_name, fields: list, values: list):
if self.__pool:
colums = ""
for i in fields:
colums = colums + i + ", "
colums = colums[:-2]
val = ""
for j in values:
val = val + "'" + str(j) + "', "
val = val[:-2]
query = "INSERT INTO {} ({}) VALUES({});".format(table_name, colums, val)
res = self.execute(query, 'insert', 'one')
return resfunc handleReq(conn net.Conn) {
buf := make([]byte, 1024)
reqLen, err := conn.Read(buf)
if err != nil {
fmt.Println("Error reading:", err.Error(), reqLen)
}
fmt.Println("\r\nRECVD: " + string(buf))
message, err := bufio.NewReader(conn).ReadString('\n')
fmt.Print("Message received:", string(message))
conn.Write([]byte("Message received."))
conn.Close()
}
