kiwi-vpn/api/kiwi_vpn_api/config.py

72 lines
1.6 KiB
Python

from enum import Enum
import json
from jose.constants import ALGORITHMS
from passlib.context import CryptContext
from pydantic import BaseModel, Field
from sqlalchemy.engine import Engine
from sqlalchemy import create_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))