This repository has been archived on 2024-05-31. You can view files and clone it, but cannot push or open issues or pull requests.
authentik/passbook/outposts/controllers/base.py

45 lines
1.2 KiB
Python
Raw Normal View History

2020-09-02 22:04:12 +00:00
"""Base Controller"""
from typing import Dict, List
2020-09-02 22:04:12 +00:00
2020-10-19 16:52:17 +00:00
from structlog import get_logger
from structlog.testing import capture_logs
2020-09-02 22:04:12 +00:00
from passbook.lib.sentry import SentryIgnoredException
2020-09-02 22:04:12 +00:00
from passbook.outposts.models import Outpost
class ControllerException(SentryIgnoredException):
"""Exception raise when anything fails during controller run"""
2020-09-02 22:04:12 +00:00
class BaseController:
"""Base Outpost deployment controller"""
deployment_ports: Dict[str, int]
outpost: Outpost
def __init__(self, outpost: Outpost):
self.outpost = outpost
2020-10-19 16:52:17 +00:00
self.logger = get_logger()
2020-09-02 22:04:12 +00:00
self.deployment_ports = {}
# pylint: disable=invalid-name
def up(self):
2020-09-02 22:04:12 +00:00
"""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 [f"{x['controller']}: {x['event']}" for x in logs]
def down(self):
"""Handler to delete everything we've created"""
raise NotImplementedError
2020-09-02 22:04:12 +00:00
def get_static_deployment(self) -> str:
"""Return a static deployment configuration"""
raise NotImplementedError