You've already forked hotpocket
114 lines
2.9 KiB
Python
114 lines
2.9 KiB
Python
# -*- coding: utf-8 -*-
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
from keep_it_secret import (
|
|
AbstractField,
|
|
EnvField,
|
|
LiteralField,
|
|
Secrets,
|
|
SecretsField,
|
|
)
|
|
|
|
|
|
class DatabaseSecrets(Secrets):
|
|
payload: str = AbstractField.new()
|
|
|
|
def get_decoded_payload(self) -> dict:
|
|
if hasattr(self, '_decoded_payload') is False:
|
|
self._decoded_payload = json.loads(self.payload)
|
|
|
|
return self._decoded_payload
|
|
|
|
@property
|
|
def engine(self) -> str:
|
|
return self.get_decoded_payload()['engine']
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return self.get_decoded_payload()['name']
|
|
|
|
@property
|
|
def user(self) -> str | None:
|
|
return self.get_decoded_payload().get('user', None)
|
|
|
|
@property
|
|
def password(self) -> str | None:
|
|
return self.get_decoded_payload().get('password', None)
|
|
|
|
@property
|
|
def host(self) -> str | None:
|
|
return self.get_decoded_payload().get('host', None)
|
|
|
|
@property
|
|
def port(self) -> str | None:
|
|
return self.get_decoded_payload().get('port', None)
|
|
|
|
|
|
class OIDCSecrets(Secrets):
|
|
DEFAULT_SCOPE = ['roles']
|
|
|
|
payload: str | None = LiteralField.new(None)
|
|
|
|
def get_decoded_payload(self) -> dict:
|
|
if hasattr(self, '_decoded_payload') is False:
|
|
if self.payload is None:
|
|
self._decoded_payload = dict()
|
|
else:
|
|
self._decoded_payload = json.loads(self.payload)
|
|
|
|
return self._decoded_payload
|
|
|
|
@property
|
|
def is_enabled(self) -> bool:
|
|
return self.payload is not None
|
|
|
|
@property
|
|
def endpoint(self) -> str | None:
|
|
return self.get_decoded_payload().get('endpoint', None)
|
|
|
|
@property
|
|
def key(self) -> str | None:
|
|
return self.get_decoded_payload().get('key', None)
|
|
|
|
@property
|
|
def secret(self) -> str | None:
|
|
return self.get_decoded_payload().get('secret', None)
|
|
|
|
@property
|
|
def scope(self) -> list[str]:
|
|
return self.get_decoded_payload().get('scope', [*self.DEFAULT_SCOPE])
|
|
|
|
@property
|
|
def display_name(self) -> str:
|
|
return self.get_decoded_payload().get('display_name', 'OIDC')
|
|
|
|
|
|
class CelerySecrets(Secrets):
|
|
broker_url: str = AbstractField.new()
|
|
result_backend: str = AbstractField.new()
|
|
|
|
|
|
class InitialAccountSecrets(Secrets):
|
|
username: str = EnvField.new(
|
|
'HOTPOCKET_BACKEND_INITIAL_ACCOUNT_USERNAME',
|
|
default=None,
|
|
required=False,
|
|
)
|
|
password: str = EnvField.new(
|
|
'HOTPOCKET_BACKEND_INITIAL_ACCOUNT_PASSWORD',
|
|
default=None,
|
|
required=False,
|
|
)
|
|
|
|
|
|
class BaseSecrets(Secrets):
|
|
SECRET_KEY: str = AbstractField.new()
|
|
|
|
DATABASE: DatabaseSecrets = SecretsField.new(DatabaseSecrets)
|
|
CELERY: CelerySecrets = SecretsField.new(CelerySecrets)
|
|
|
|
OIDC: OIDCSecrets = SecretsField.new(OIDCSecrets)
|
|
INITIAL_ACCOUNT: InitialAccountSecrets = SecretsField.new(InitialAccountSecrets)
|