Source code for betty.extension.nginx.docker

"""
Integrate Betty with Docker.
"""

import asyncio
from pathlib import Path
from types import TracebackType

import docker
from docker.models.containers import Container as DockerContainer



[docs]
class Container:
    _IMAGE_TAG = "betty-serve"

    def __init__(
        self,
        www_directory_path: Path,
        docker_directory_path: Path,
        nginx_configuration_file_path: Path,
    ):
        self._docker_directory_path = docker_directory_path
        self._nginx_configuration_file_path = nginx_configuration_file_path
        self._www_directory_path = www_directory_path
        self._client = docker.from_env()
        self.__container: DockerContainer | None = None

    async def __aenter__(self) -> None:
        await self.start()

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
        await self.stop()


[docs]
    async def start(self) -> None:
        await asyncio.to_thread(self._start)


    def _start(self) -> None:
        self._client.images.build(
            path=str(self._docker_directory_path), tag=self._IMAGE_TAG
        )
        self._container.start()
        self._container.exec_run(["nginx", "-s", "reload"])


[docs]
    async def stop(self) -> None:
        await asyncio.to_thread(self._stop)


    def _stop(self) -> None:
        if self._container is not None:
            self._container.stop()

    @property
    def _container(self) -> DockerContainer:
        if self.__container is None:
            self.__container = self._client.containers.create(
                self._IMAGE_TAG,
                auto_remove=True,
                detach=True,
                volumes={
                    self._nginx_configuration_file_path: {
                        "bind": "/etc/nginx/conf.d/betty.conf",
                        "mode": "ro",
                    },
                    self._www_directory_path: {
                        "bind": "/var/www/betty",
                        "mode": "ro",
                    },
                },
            )
        return self.__container

    @property
    def ip(self) -> DockerContainer:
        return self._client.api.inspect_container(self._container.id)[
            "NetworkSettings"
        ]["Networks"]["bridge"]["IPAddress"]