-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathdb_config.py
41 lines (31 loc) · 1.03 KB
/
db_config.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
import aioredis
from pydantic import BaseSettings, RedisDsn
from peewee import PostgresqlDatabase
class Database(BaseSettings):
POSTGRES_DB: str
POSTGRES_HOST: str
POSTGRES_PORT: int
POSTGRES_USER: str
POSTGRES_PASSWORD: str
REDIS_DB: RedisDsn = "redis://127.0.0.1:6379/0"
def get_db_name(self):
return f"postgres://{self.POSTGRES_USER}:{self.POSTGRES_PASSWORD}@{self.POSTGRES_HOST}:{self.POSTGRES_PORT}/{self.POSTGRES_DB}"
def get_db_dict(self) -> dict:
return dict(
database=self.POSTGRES_DB,
user=self.POSTGRES_USER,
password=self.POSTGRES_PASSWORD,
host=self.POSTGRES_HOST,
port=self.POSTGRES_PORT
)
db_config = Database(
_env_file='.env',
_env_file_encoding='utf-8'
)
# # redis init
REDIS_CLIENT = aioredis.from_url(url=db_config.REDIS_DB, encoding="utf-8", decode_responses=True)
def psql():
db = PostgresqlDatabase(**db_config.get_db_dict())
db.connect()
return db
db: PostgresqlDatabase = psql()