hotpocket/services/backend/hotpocket_backend/secrets/base.py
Tomek Wójcik b4338e2769
Some checks failed
CI / Checks (push) Failing after 13m2s
Release v1.0.0
2025-08-20 21:00:50 +02:00

94 lines
2.5 KiB
Python

# -*- coding: utf-8 -*-
from __future__ import annotations
import json
from keep_it_secret import AbstractField, 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 BaseSecrets(Secrets):
SECRET_KEY: str = AbstractField.new()
DATABASE: DatabaseSecrets = SecretsField.new(DatabaseSecrets)
CELERY: CelerySecrets = SecretsField.new(CelerySecrets)
OIDC: OIDCSecrets = SecretsField.new(OIDCSecrets)