bc9e7e8b93
* build(deps): bump structlog from 20.1.0 to 20.2.0 Bumps [structlog](https://github.com/hynek/structlog) from 20.1.0 to 20.2.0. - [Release notes](https://github.com/hynek/structlog/releases) - [Changelog](https://github.com/hynek/structlog/blob/master/CHANGELOG.rst) - [Commits](https://github.com/hynek/structlog/compare/20.1.0...20.2.0) Signed-off-by: dependabot[bot] <support@github.com> * *: use structlog.stdlib instead of structlog for type-hints Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: Jens Langhammer <jens.langhammer@beryju.org>
58 lines
1.5 KiB
Python
58 lines
1.5 KiB
Python
"""Base Controller"""
|
|
from dataclasses import dataclass
|
|
|
|
from structlog.stdlib import get_logger
|
|
from structlog.testing import capture_logs
|
|
|
|
from authentik.lib.sentry import SentryIgnoredException
|
|
from authentik.outposts.models import Outpost, OutpostServiceConnection
|
|
|
|
FIELD_MANAGER = "goauthentik.io"
|
|
|
|
|
|
class ControllerException(SentryIgnoredException):
|
|
"""Exception raised when anything fails during controller run"""
|
|
|
|
|
|
@dataclass
|
|
class DeploymentPort:
|
|
"""Info about deployment's single port."""
|
|
|
|
port: int
|
|
name: str
|
|
protocol: str
|
|
|
|
|
|
class BaseController:
|
|
"""Base Outpost deployment controller"""
|
|
|
|
deployment_ports: list[DeploymentPort]
|
|
|
|
outpost: Outpost
|
|
connection: OutpostServiceConnection
|
|
|
|
def __init__(self, outpost: Outpost, connection: OutpostServiceConnection):
|
|
self.outpost = outpost
|
|
self.connection = connection
|
|
self.logger = get_logger()
|
|
self.deployment_ports = []
|
|
|
|
# pylint: disable=invalid-name
|
|
def up(self):
|
|
"""Called by scheduled task to reconcile deployment/service/etc"""
|
|
raise NotImplementedError
|
|
|
|
def up_with_logs(self) -> list[str]:
|
|
"""Call .up() but capture all log output and return it."""
|
|
with capture_logs() as logs:
|
|
self.up()
|
|
return [x["event"] for x in logs]
|
|
|
|
def down(self):
|
|
"""Handler to delete everything we've created"""
|
|
raise NotImplementedError
|
|
|
|
def get_static_deployment(self) -> str:
|
|
"""Return a static deployment configuration"""
|
|
raise NotImplementedError
|