python-gRPC

1
from dataclasses import dataclass
2
from os import getenv
3
from sqlalchemy.engine import URL
4
from dotenv import load_dotenv
5

6
load_dotenv()
7

8

9
@dataclass
10
class DataBaseConfig:
11
    name_db: str | None = getenv("POSTGRES_DB_NAME")
12
    user: str | None = getenv("POSTGRES_USERNAME")
13
    password: str | None = getenv("POSTGRES_PASSWORD")
14
    port: str | None = getenv("POSTGRES_PORT")
15
    host: str | None = getenv("POSTGRES_HOST")
16

17
    driver: str = "asyncpg"
18
    database_system = "postgresql"
19

20
    def __post_init__(self):
21
        required_vars = ["name_db", "user", "password", "port", "host"]
22
        for var in required_vars:
23
            if getattr(self, var) is None:
24
                raise ValueError(
25
                    f"Нет переменной {var} для коннекта БД в окружении проекта."
26
                )
27

28
    def build_connection_str(self) -> str:
29
        return URL.create(
30
            drivername=f"{self.database_system}+{self.driver}",
31
            username=self.user,
32
            password=self.password,
33
            host=self.host,
34
            port=self.port,
35
            database=self.name_db,
36
        ).render_as_string(hide_password=False)
37

38

39
def get_connection_params():
40
    return {
41
        "database": getenv("POSTGRES_DB_NAME", None),
42
        "user": getenv("POSTGRES_USERNAME", None),
43
        "password": getenv("POSTGRES_PASSWORD", None),
44
        "host": getenv("POSTGRES_HOST", "localhost"),
45
        "port": getenv("POSTGRES_PORT", "5432"),
46
    }
47