Source code for runway.cfngin.hooks.awslambda.python_requirements._docker

"""Docker logic for python."""
from __future__ import annotations

import logging
import re
from typing import TYPE_CHECKING, Dict, List, Optional, Union

from docker.types.services import Mount

from .....compat import cached_property, shlex_join
from .....utils import Version
from ..docker import DockerDependencyInstaller

if TYPE_CHECKING:
    from docker.client import DockerClient

    from .....context import CfnginContext, RunwayContext
    from . import PythonProject


[docs]class PythonDockerDependencyInstaller(DockerDependencyInstaller): """Docker dependency installer for Python.""" project: PythonProject
[docs] def __init__( self, project: PythonProject, *, client: Optional[DockerClient] = None, context: Optional[Union[CfnginContext, RunwayContext]] = None, ) -> None: """Instantiate class. Args: project: awslambda project. client: Pre-configured :class:`docker.client.DockerClient`. context: CFNgin or Runway context object. """ super().__init__(project, client=client, context=context)
@cached_property def bind_mounts(self) -> List[Mount]: """Bind mounts that will be used by the container.""" mounts = [*super().bind_mounts] if self.project.requirements_txt: mounts.append( Mount( target=f"/var/task/{self.project.requirements_txt.name}", source=str(self.project.requirements_txt), type="bind", ) ) return mounts @cached_property def environment_variables(self) -> Dict[str, str]: """Environment variables to pass to the docker container. This is a subset of the environment variables stored in the context object as some will cause issues if they are passed. """ docker_env_vars = super().environment_variables pip_env_vars = { k: v for k, v in self.ctx.env.vars.items() if k.startswith("PIP") } return {**docker_env_vars, **pip_env_vars} @cached_property def install_commands(self) -> List[str]: """Commands to run to install dependencies.""" if self.project.requirements_txt: return [ shlex_join( self.project.pip.generate_install_command( cache_dir=self.CACHE_DIR if self.project.cache_dir else None, no_cache_dir=not self.project.args.use_cache, no_deps=bool(self.project.poetry or self.project.pipenv), requirements=f"/var/task/{self.project.requirements_txt.name}", target=self.DEPENDENCY_DIR, ) + (self.project.args.extend_pip_args or []) ) ] return [] @cached_property def python_version(self) -> Optional[Version]: """Version of Python installed in the docker container.""" match = re.search( r"Python (?P<version>\S*)", "\n".join(self.run_command("python --version", level=logging.DEBUG)), ) if not match: return None return Version(match.group("version")) @cached_property def runtime(self) -> Optional[str]: """AWS Lambda runtime determined from the docker container's Python version.""" if not self.python_version: return None return f"python{self.python_version.major}.{self.python_version.minor}"