kiwi-vpn/api/kiwi_vpn_api/config.py

85 lines
2.1 KiB
Python

from __future__ import annotations
import functools
import json
from enum import Enum
from jose.constants import ALGORITHMS
from passlib.context import CryptContext
from pydantic import BaseModel, BaseSettings, Field
from sqlalchemy import create_engine
from sqlalchemy.engine import Engine
class Settings(BaseSettings):
production_mode: bool = False
config_file: str = "tmp/config.json"
openapi_url: str = "/openapi.json"
docs_url: str | None = "/docs"
redoc_url: str | None = None
@staticmethod
@functools.lru_cache
def get() -> Settings:
return Settings()
class DBType(Enum):
sqlite = "sqlite"
mysql = "mysql"
class DBConfig(BaseModel):
db_type: DBType = DBType.sqlite
@property
def db_engine(self) -> Engine:
return create_engine(
"sqlite:///./tmp/vpn.db",
connect_args={"check_same_thread": False},
)
class JWTConfig(BaseModel):
secret: str | None = None
hash_algorithm: str = ALGORITHMS.HS256
expiry_minutes: int = 30
class CryptoConfig(BaseModel):
schemes: list[str] = ["bcrypt"]
@property
def crypt_context(self) -> CryptContext:
return CryptContext(
schemes=self.schemes,
deprecated="auto",
)
class Config(BaseModel):
db: DBConfig = Field(default_factory=DBConfig)
jwt: JWTConfig = Field(default_factory=JWTConfig)
crypto: CryptoConfig = Field(default_factory=CryptoConfig)
@staticmethod
async def get() -> Config | None:
try:
with open(Settings.get().config_file, "r") as config_file:
return Config.parse_obj(json.load(config_file))
except FileNotFoundError:
return None
@staticmethod
def set(config: Config) -> None:
with open(Settings.get().config_file, "w") as config_file:
config_file.write(config.json(indent=2))
@property
def crypt_context(self) -> CryptContext:
return self.crypto.crypt_context
@property
def db_engine(self) -> Engine:
return self.db.db_engine