def METHOD_NAME(self, run, docker_image, command):
    container_context = self.get_container_context(run)
    docker_env = dict([parse_env_var(env_var) for env_var in container_context.env_vars])
    docker_env["DAGSTER_RUN_JOB_NAME"] = run.job_name
    client = self._get_client(container_context)
    try:
        container = client.containers.create(
            image=docker_image,
            command=command,
            detach=True,
            environment=docker_env,
            network=container_context.networks[0] if len(container_context.networks) else None,
            **container_context.container_kwargs,
        )
    except docker.errors.ImageNotFound:
        client.images.pull(docker_image)
        container = client.containers.create(
            image=docker_image,
            command=command,
            detach=True,
            environment=docker_env,
            network=container_context.networks[0] if len(container_context.networks) else None,
            **container_context.container_kwargs,
        )
    if len(container_context.networks) > 1:
        for network_name in container_context.networks[1:]:
            network = client.networks.get(network_name)
            network.connect(container)
    self._instance.report_engine_event(
        message=(
            "Launching run in a new container {container_id} with image {docker_image}".format(
                container_id=container.id,
                docker_image=docker_image,
            )
        ),
        dagster_run=run,
        cls=self.__class__,
    )
    self._instance.add_run_tags(
        run.run_id,
        {DOCKER_CONTAINER_ID_TAG: container.id, DOCKER_IMAGE_TAG: docker_image},
    )
    container.start()