import psycopg2.pool
from psycopg2 import errors
dbname = 'mydatabase'
user = 'postgres'
password = '123'
host = 'localhost'
port = '5432'
connection_string = f"dbname={dbname} user={user} password={password} host={host} port={port}"
class DatabaseConnection:
def __init__(self, connection_string):
self._connection_string = connection_string
self._connection_pool = psycopg2.pool.SimpleConnectionPool(minconn=1, maxconn=10, dsn=connection_string)
self._connection = None
def __enter__(self):
self._connection = self._connection_pool.getconn()
return self._connection.cursor()
def __exit__(self, exc_type, exc_val, exc_tb):
if self._connection:
self._connection.commit()
self._connection_pool.putconn(self._connection)
class PostgreSQLDatabase:
def __init__(self, connection_string):
self._connection_string = connection_string
def __enter__(self):
self._connection = DatabaseConnection(self._connection_string)
return self._connection.__enter__()
def __exit__(self, exc_type, exc_val, exc_tb):
return self._connection.__exit__(exc_type, exc_val, exc_tb)
class Admin_connected:
def __init__(self):
self.database = PostgreSQLDatabase(connection_string)
def create_database_if_not_exists(self):
# Подключаемся к базе postgres, чтобы узнать, существует ли база
conn = psycopg2.connect(database='postgres', user=user, password=password, host=host,
port=port)
conn.autocommit = True
cursor = conn.cursor()
# Проверка наличия базы данных
cursor.execute("SELECT 1 FROM pg_database WHERE datname = %s", (dbname,))
exists = cursor.fetchone()
if not exists:
try:
# Создание новой базы данных
with conn.cursor() as cursor:
cursor.execute(f"CREATE DATABASE {dbname}")
print(f"База данных '{dbname}' создана.")
self.create_user_table()
except Exception as e:
print(f'Ошибка {e}')
finally:
if conn:
conn.close()
else:
print(f"База данных '{dbname}' уже существует.")
def create_user_table(self):
try:
with self.database as cursor:
cursor.execute(f'CREATE TABLE IF NOT EXISTS public.users '
f"(id integer NOT NULL, user_id integer NOT NULL, PRIMARY KEY (id));"
f'ALTER TABLE IF EXISTS public.users OWNER to postgres;')
print(f"[INFO] Таблица пользователей успешно создана")
except Exception as e:
print(f'Ошибка при создании новой таблицы: {e}')
def exists(self):
"""Проверяем на наличие в базе данных"""
try:
with self.database as cursor:
cursor.execute("SELECT * FROM users")
print("[INFO] Данные успешно получены")
return cursor.fetchall()
except errors.OperationalError:
self.create_database_if_not_exists()
return self.exists()
db = Admin_connected()
print(db.exists())class Client_connect:
def __init__(self):
self.dbname = os.getenv('dbname')
self.user = os.getenv('user')
self.password = os.getenv('password')
self.host = os.getenv('host')
def __call__(self):
try:
self.conn = psycopg2.connect(dbname=self.dbname,
user=self.user,
password=self.password,
host=self.host)
print("[INFO] подключение к PostgreSQL")
except Exception as e:
print(f"[ERROR - def __call__] Ошибка получения доступа к базе данных: {e}")
def user_exists(self, user_id):
Client_connect.__call__(self)
"""Проверяем на наличие юзера в базе данных"""
try:
with self.conn.cursor() as cursor:
cursor.execute("SELECT id FROM users WHERE tele_user_id = %s", (user_id, ))
print("[INFO] Данные успешно получены")
return bool(len(cursor.fetchall()))
except Exception as e:
print(f"[ERROR] Ошибка работы базы данных: {e}")
finally:
if self.conn:
self.conn.close()
print("[INFO] PostgreSQL соединение закрыто")
Связка прослушки одного порта
server {
listen 12345;
server_name _;
return 444;
}
##
## HTTPS + mTLS
##
server {
listen 12345 ssl;
server_name _;
Вызывает ошибку, без назначения default_server. Но при назначении такового на что-либо, начинает игнорироваться другой сервер, в итоге мы видим ту же самую картину, когда на обычный http порт прилетает заглушка от nginx. Как я понимаю, она прилетает ещё до входа в server {}, по этому, конфигом это не получается отредактировать...