47 lines
1.8 KiB
Python
47 lines
1.8 KiB
Python
# Database configuration package
|
||
from pydantic import BaseModel
|
||
import re
|
||
import os
|
||
import urllib.parse
|
||
|
||
class Settings(BaseModel):
|
||
DB_HOST: str = os.getenv("DB_HOST", "postgres")
|
||
DB_PORT: int = int(os.getenv("DB_PORT", "5432"))
|
||
DB_USER: str = os.getenv("DB_USER", "postgres")
|
||
DB_PASS: str = os.getenv("DB_PASS", "")
|
||
DB_NAME: str = os.getenv("DB_NAME", "profi")
|
||
|
||
@classmethod
|
||
def validate_db_port(cls, v):
|
||
# Проверка и преобразование DB_PORT
|
||
if isinstance(v, str):
|
||
# Используем регулярное выражение для извлечения целого числа
|
||
match = re.search(r'\d+', v)
|
||
if match:
|
||
return int(match.group())
|
||
else:
|
||
raise ValueError(f"Invalid DB_PORT value: {v}")
|
||
return v
|
||
|
||
@property
|
||
def DATABASE_URL_asyncpg(self) -> str:
|
||
pwd = urllib.parse.quote_plus(str(self.DB_PASS))
|
||
return f"postgresql+asyncpg://{self.DB_USER}:{pwd}@{self.DB_HOST}:{self.DB_PORT}/{self.DB_NAME}"
|
||
|
||
@property
|
||
def DATABASE_URL_syncpg(self) -> str:
|
||
pwd = urllib.parse.quote_plus(str(self.DB_PASS))
|
||
return f"postgresql+pg8000://{self.DB_USER}:{pwd}@{self.DB_HOST}:{self.DB_PORT}/{self.DB_NAME}"
|
||
|
||
@property
|
||
def PG8000_CONNECT_ARGS(self) -> dict:
|
||
"""connect_args для pg8000 — отключаем SSL (сервер его не поддерживает)."""
|
||
return {"ssl_context": None}
|
||
|
||
|
||
def make_engine():
|
||
"""Создаёт SQLAlchemy engine с правильными параметрами для pg8000."""
|
||
from sqlalchemy import create_engine
|
||
s = Settings()
|
||
return create_engine(s.DATABASE_URL_syncpg, connect_args=s.PG8000_CONNECT_ARGS, future=True)
|