kiwi-vpn/api/kiwi_vpn_api/config.py

import json
from enum import Enum

from jose.constants import ALGORITHMS
from passlib.context import CryptContext
from pydantic import BaseModel, Field
from sqlalchemy import create_engine
from sqlalchemy.engine import Engine

CONFIG_FILE = "tmp/config.json"


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 BaseConfig(BaseModel):
    db: DBConfig = Field(default_factory=DBConfig)
    jwt: JWTConfig = Field(default_factory=JWTConfig)
    crypto: CryptoConfig = Field(default_factory=CryptoConfig)

    @property
    def crypt_context(self) -> CryptContext:
        return self.crypto.crypt_context

    @property
    def db_engine(self) -> Engine:
        return self.db.db_engine


async def get() -> BaseConfig | None:
    try:
        with open(CONFIG_FILE, "r") as config_file:
            return BaseConfig.parse_obj(json.load(config_file))

    except FileNotFoundError:
        return None


def set(config: BaseConfig) -> None:
    with open(CONFIG_FILE, "w") as config_file:
        config_file.write(config.json(indent=2))