db / app.py
aigenai's picture
Update app.py
130ce2e verified
from flask import Flask, jsonify
import psycopg2
import os
import socket
app = Flask(__name__)
def check_port(host: str, port: int):
"""Check if a specific port on a host is open."""
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
sock.settimeout(1) # 1 second timeout
try:
sock.connect((host, port))
return True
except socket.error:
return False
def connect_db():
"""Attempt to connect to the PostgreSQL database and retrieve version info."""
try:
connection = psycopg2.connect(
dbname=os.getenv('POSTGRES_DB'),
user=os.getenv('POSTGRES_USER'),
password=os.getenv('POSTGRES_PASSWORD'),
host='localhost'
)
cursor = connection.cursor()
cursor.execute("SELECT version();")
db_version = cursor.fetchone()
return {
"status": "success",
"db_version": db_version,
"user": os.getenv('POSTGRES_USER'),
"database": os.getenv('POSTGRES_DB')
}
except Exception as e:
return {"status": "failure", "error": str(e)}
@app.route('/')
def home():
# Check if PostgreSQL is running and port is open
db_running = check_port('localhost', 5432)
# Perform database connection test
db_info = connect_db()
result = {
"message": "Database Connection Test Results",
"database_running": db_running,
"port_open": db_running, # The same check, for clarity
"connection_result": db_info
}
return jsonify(result)
if __name__ == '__main__':
app.run(host='0.0.0.0', port=7860)