import os
from http.server import BaseHTTPRequestHandler, HTTPServer
import psycopg2
from psycopg2 import pool
# Database connection parameters
DB_USER = os.getenv('DB_USER')
DB_PASSWORD = os.getenv('DB_PASSWORD')
DB_HOST = os.getenv('DB_HOST')
DB_PORT = os.getenv('DB_PORT')
DB_NAME = os.getenv('DB_NAME')
DATABASE_URI = f"postgresql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}"
# Initialize the connection pool
connection_pool = None
class RequestHandler(BaseHTTPRequestHandler):
def do_GET(self):
global connection_pool
if not connection_pool:
try:
# Create the connection pool
connection_pool = pool.SimpleConnectionPool(
minconn=1, maxconn=5, dsn=DATABASE_URI
)
response = "Connection pool created successfully."
except Exception as error:
response = f"Error creating connection pool: {error}"
self.respond(response)
return
try:
# Obtain a connection from the pool
pooled_conn = connection_pool.getconn()
cursor = pooled_conn.cursor()
cursor.execute("SELECT version();")
db_version = cursor.fetchone()
response = f"Pooling successful, PostgreSQL version: {db_version}"
connection_pool.putconn(pooled_conn) # Return connection to the pool
except Exception as error:
response = f"Error connecting with pooling: {error}"
# Send response to client
self.respond(response)
def respond(self, message):
self.send_response(200)
self.send_header("Content-type", "text/html")
self.end_headers()
self.wfile.write(f"<html><body><h1>{message}</h1></body></html>".encode("utf-8"))
def run(server_class=HTTPServer, handler_class=RequestHandler):
server_address = ('', 8080) # Serve on port 8080
httpd = server_class(server_address, handler_class)
print("Server running on port 8080")
httpd.serve_forever()
if __name__ == "__main__":
run()