# Code generated by dagger. DO NOT EDIT.
import warnings
from collections.abc import Callable
from dataclasses import dataclass
from typing_extensions import Self
from dagger.client._core import Arg, Root
from dagger.client._guards import typecheck
from dagger.client.base import Enum, Input, Scalar, Type
class CacheVolumeID(Scalar):
"""The `CacheVolumeID` scalar type represents an identifier for an
object of type CacheVolume."""
class ContainerID(Scalar):
"""The `ContainerID` scalar type represents an identifier for an
object of type Container."""
class CurrentModuleID(Scalar):
"""The `CurrentModuleID` scalar type represents an identifier for an
object of type CurrentModule."""
class DirectoryID(Scalar):
"""The `DirectoryID` scalar type represents an identifier for an
object of type Directory."""
class EngineCacheEntryID(Scalar):
"""The `EngineCacheEntryID` scalar type represents an identifier for
an object of type EngineCacheEntry."""
class EngineCacheEntrySetID(Scalar):
"""The `EngineCacheEntrySetID` scalar type represents an identifier
for an object of type EngineCacheEntrySet."""
class EngineCacheID(Scalar):
"""The `EngineCacheID` scalar type represents an identifier for an
object of type EngineCache."""
class EngineID(Scalar):
"""The `EngineID` scalar type represents an identifier for an object
of type Engine."""
class EnumTypeDefID(Scalar):
"""The `EnumTypeDefID` scalar type represents an identifier for an
object of type EnumTypeDef."""
class EnumValueTypeDefID(Scalar):
"""The `EnumValueTypeDefID` scalar type represents an identifier for
an object of type EnumValueTypeDef."""
class EnvVariableID(Scalar):
"""The `EnvVariableID` scalar type represents an identifier for an
object of type EnvVariable."""
class ErrorID(Scalar):
"""The `ErrorID` scalar type represents an identifier for an object of
type Error."""
class FieldTypeDefID(Scalar):
"""The `FieldTypeDefID` scalar type represents an identifier for an
object of type FieldTypeDef."""
class FileID(Scalar):
"""The `FileID` scalar type represents an identifier for an object of
type File."""
class FunctionArgID(Scalar):
"""The `FunctionArgID` scalar type represents an identifier for an
object of type FunctionArg."""
class FunctionCallArgValueID(Scalar):
"""The `FunctionCallArgValueID` scalar type represents an identifier
for an object of type FunctionCallArgValue."""
class FunctionCallID(Scalar):
"""The `FunctionCallID` scalar type represents an identifier for an
object of type FunctionCall."""
class FunctionID(Scalar):
"""The `FunctionID` scalar type represents an identifier for an object
of type Function."""
class GeneratedCodeID(Scalar):
"""The `GeneratedCodeID` scalar type represents an identifier for an
object of type GeneratedCode."""
class GitModuleSourceID(Scalar):
"""The `GitModuleSourceID` scalar type represents an identifier for an
object of type GitModuleSource."""
class GitRefID(Scalar):
"""The `GitRefID` scalar type represents an identifier for an object
of type GitRef."""
class GitRepositoryID(Scalar):
"""The `GitRepositoryID` scalar type represents an identifier for an
object of type GitRepository."""
class HostID(Scalar):
"""The `HostID` scalar type represents an identifier for an object of
type Host."""
class InputTypeDefID(Scalar):
"""The `InputTypeDefID` scalar type represents an identifier for an
object of type InputTypeDef."""
class InterfaceTypeDefID(Scalar):
"""The `InterfaceTypeDefID` scalar type represents an identifier for
an object of type InterfaceTypeDef."""
class JSON(Scalar):
"""An arbitrary JSON-encoded value."""
class LabelID(Scalar):
"""The `LabelID` scalar type represents an identifier for an object of
type Label."""
class ListTypeDefID(Scalar):
"""The `ListTypeDefID` scalar type represents an identifier for an
object of type ListTypeDef."""
class LocalModuleSourceID(Scalar):
"""The `LocalModuleSourceID` scalar type represents an identifier for
an object of type LocalModuleSource."""
class ModuleDependencyID(Scalar):
"""The `ModuleDependencyID` scalar type represents an identifier for
an object of type ModuleDependency."""
class ModuleID(Scalar):
"""The `ModuleID` scalar type represents an identifier for an object
of type Module."""
class ModuleSourceID(Scalar):
"""The `ModuleSourceID` scalar type represents an identifier for an
object of type ModuleSource."""
class ModuleSourceViewID(Scalar):
"""The `ModuleSourceViewID` scalar type represents an identifier for
an object of type ModuleSourceView."""
class ObjectTypeDefID(Scalar):
"""The `ObjectTypeDefID` scalar type represents an identifier for an
object of type ObjectTypeDef."""
class Platform(Scalar):
"""The platform config OS and architecture in a Container. The format
is [os]/[platform]/[version] (e.g., "darwin/arm64/v7",
"windows/amd64", "linux/arm64")."""
class PortID(Scalar):
"""The `PortID` scalar type represents an identifier for an object of
type Port."""
class ScalarTypeDefID(Scalar):
"""The `ScalarTypeDefID` scalar type represents an identifier for an
object of type ScalarTypeDef."""
class SecretID(Scalar):
"""The `SecretID` scalar type represents an identifier for an object
of type Secret."""
class ServiceID(Scalar):
"""The `ServiceID` scalar type represents an identifier for an object
of type Service."""
class SocketID(Scalar):
"""The `SocketID` scalar type represents an identifier for an object
of type Socket."""
class SourceMapID(Scalar):
"""The `SourceMapID` scalar type represents an identifier for an
object of type SourceMap."""
class TerminalID(Scalar):
"""The `TerminalID` scalar type represents an identifier for an object
of type Terminal."""
class TypeDefID(Scalar):
"""The `TypeDefID` scalar type represents an identifier for an object
of type TypeDef."""
class Void(Scalar):
"""The absence of a value. A Null Void is used as a placeholder for
resolvers that do not return anything."""
class CacheSharingMode(Enum):
"""Sharing mode of the cache volume."""
LOCKED = "LOCKED"
"""Shares the cache volume amongst many build pipelines, but will serialize the writes"""
PRIVATE = "PRIVATE"
"""Keeps a cache volume for a single build pipeline"""
SHARED = "SHARED"
"""Shares the cache volume amongst many build pipelines"""
class ImageLayerCompression(Enum):
"""Compression algorithm to use for image layers."""
EStarGZ = "EStarGZ"
Gzip = "Gzip"
Uncompressed = "Uncompressed"
Zstd = "Zstd"
class ImageMediaTypes(Enum):
"""Mediatypes to use in published or exported image metadata."""
DockerMediaTypes = "DockerMediaTypes"
OCIMediaTypes = "OCIMediaTypes"
class ModuleSourceKind(Enum):
"""The kind of module source."""
GIT_SOURCE = "GIT_SOURCE"
LOCAL_SOURCE = "LOCAL_SOURCE"
class NetworkProtocol(Enum):
"""Transport layer network protocol associated to a port."""
TCP = "TCP"
UDP = "UDP"
class ReturnType(Enum):
"""Expected return type of an execution"""
ANY = "ANY"
"""Any execution (exit codes 0-127)"""
FAILURE = "FAILURE"
"""A failed execution (exit codes 1-127)"""
SUCCESS = "SUCCESS"
"""A successful execution (exit code 0)"""
class TypeDefKind(Enum):
"""Distinguishes the different kinds of TypeDefs."""
BOOLEAN_KIND = "BOOLEAN_KIND"
"""A boolean value."""
ENUM_KIND = "ENUM_KIND"
"""A GraphQL enum type and its values
Always paired with an EnumTypeDef.
"""
INPUT_KIND = "INPUT_KIND"
"""A graphql input type, used only when representing the core API via TypeDefs."""
INTEGER_KIND = "INTEGER_KIND"
"""An integer value."""
INTERFACE_KIND = "INTERFACE_KIND"
"""A named type of functions that can be matched+implemented by other objects+interfaces.
Always paired with an InterfaceTypeDef.
"""
LIST_KIND = "LIST_KIND"
"""A list of values all having the same type.
Always paired with a ListTypeDef.
"""
OBJECT_KIND = "OBJECT_KIND"
"""A named type defined in the GraphQL schema, with fields and functions.
Always paired with an ObjectTypeDef.
"""
SCALAR_KIND = "SCALAR_KIND"
"""A scalar value of any basic kind."""
STRING_KIND = "STRING_KIND"
"""A string value."""
VOID_KIND = "VOID_KIND"
"""A special kind used to signify that no value is returned.
This is used for functions that have no return value. The outer TypeDef specifying this Kind is always Optional, as the Void is never actually represented.
"""
@typecheck
@dataclass(slots=True)
class BuildArg(Input):
"""Key value object that represents a build argument."""
name: str
"""The build argument name."""
value: str
"""The build argument value."""
@typecheck
@dataclass(slots=True)
class PipelineLabel(Input):
"""Key value object that represents a pipeline label."""
name: str
"""Label name."""
value: str
"""Label value."""
@typecheck
@dataclass(slots=True)
class PortForward(Input):
"""Port forwarding rules for tunneling network traffic."""
backend: int
"""Destination port for traffic."""
frontend: int | None = None
"""Port to expose to clients. If unspecified, a default will be chosen."""
protocol: NetworkProtocol | None = NetworkProtocol.TCP
"""Transport layer protocol to use for traffic."""
@typecheck
class CacheVolume(Type):
"""A directory whose contents persist across runs."""
[docs]
async def id(self) -> CacheVolumeID:
"""A unique identifier for this CacheVolume.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
CacheVolumeID
The `CacheVolumeID` scalar type represents an identifier for an
object of type CacheVolume.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(CacheVolumeID)
@typecheck
class Container(Type):
"""An OCI-compatible container, also known as a Docker container."""
[docs]
def as_service(
self,
*,
args: list[str] | None = None,
use_entrypoint: bool | None = False,
experimental_privileged_nesting: bool | None = False,
insecure_root_capabilities: bool | None = False,
expand: bool | None = False,
no_init: bool | None = False,
) -> "Service":
"""Turn the container into a Service.
Be sure to set any exposed ports before this conversion.
Parameters
----------
args:
Command to run instead of the container's default command (e.g.,
["go", "run", "main.go"]).
If empty, the container's default command is used.
use_entrypoint:
If the container has an entrypoint, prepend it to the args.
experimental_privileged_nesting:
Provides Dagger access to the executed command.
Do not use this option unless you trust the command being
executed; the command being executed WILL BE GRANTED FULL ACCESS
TO YOUR HOST FILESYSTEM.
insecure_root_capabilities:
Execute the command with all root capabilities. This is similar to
running a command with "sudo" or executing "docker run" with the "
--privileged" flag. Containerization does not provide any security
guarantees when using this option. It should only be used when
absolutely necessary and only with trusted commands.
expand:
Replace "${VAR}" or "$VAR" in the args according to the current
environment variables defined in the container (e.g. "/$VAR/foo").
no_init:
If set, skip the automatic init process injected into containers
by default.
This should only be used if the user requires that their exec
process be the pid 1 process in the container. Otherwise it may
result in unexpected behavior.
"""
_args = [
Arg("args", () if args is None else args, ()),
Arg("useEntrypoint", use_entrypoint, False),
Arg(
"experimentalPrivilegedNesting", experimental_privileged_nesting, False
),
Arg("insecureRootCapabilities", insecure_root_capabilities, False),
Arg("expand", expand, False),
Arg("noInit", no_init, False),
]
_ctx = self._select("asService", _args)
return Service(_ctx)
[docs]
def as_tarball(
self,
*,
platform_variants: "list[Container] | None" = None,
forced_compression: ImageLayerCompression | None = None,
media_types: ImageMediaTypes | None = ImageMediaTypes.OCIMediaTypes,
) -> "File":
"""Returns a File representing the container serialized to a tarball.
Parameters
----------
platform_variants:
Identifiers for other platform specific containers.
Used for multi-platform images.
forced_compression:
Force each layer of the image to use the specified compression
algorithm.
If this is unset, then if a layer already has a compressed blob in
the engine's cache, that will be used (this can result in a mix of
compression algorithms for different layers). If this is unset and
a layer has no compressed blob in the engine's cache, then it will
be compressed using Gzip.
media_types:
Use the specified media types for the image's layers.
Defaults to OCI, which is largely compatible with most recent
container runtimes, but Docker may be needed for older runtimes
without OCI support.
"""
_args = [
Arg(
"platformVariants",
() if platform_variants is None else platform_variants,
(),
),
Arg("forcedCompression", forced_compression, None),
Arg("mediaTypes", media_types, ImageMediaTypes.OCIMediaTypes),
]
_ctx = self._select("asTarball", _args)
return File(_ctx)
[docs]
def build(
self,
context: "Directory",
*,
dockerfile: str | None = "Dockerfile",
target: str | None = "",
build_args: list[BuildArg] | None = None,
secrets: "list[Secret] | None" = None,
) -> Self:
"""Initializes this container from a Dockerfile build.
Parameters
----------
context:
Directory context used by the Dockerfile.
dockerfile:
Path to the Dockerfile to use.
target:
Target build stage to build.
build_args:
Additional build arguments.
secrets:
Secrets to pass to the build.
They will be mounted at /run/secrets/[secret-name] in the build
container
They can be accessed in the Dockerfile using the "secret" mount
type and mount path /run/secrets/[secret-name], e.g. RUN
--mount=type=secret,id=my-secret curl
[http://example.com?token=$(cat /run/secrets/my-
secret)](http://example.com?token=$(cat /run/secrets/my-secret))
"""
_args = [
Arg("context", context),
Arg("dockerfile", dockerfile, "Dockerfile"),
Arg("target", target, ""),
Arg("buildArgs", () if build_args is None else build_args, ()),
Arg("secrets", () if secrets is None else secrets, ()),
]
_ctx = self._select("build", _args)
return Container(_ctx)
[docs]
async def default_args(self) -> list[str]:
"""Retrieves default arguments for future commands.
Returns
-------
list[str]
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("defaultArgs", _args)
return await _ctx.execute(list[str])
[docs]
def directory(
self,
path: str,
*,
expand: bool | None = False,
) -> "Directory":
"""Retrieves a directory at the given path.
Mounts are included.
Parameters
----------
path:
The path of the directory to retrieve (e.g., "./src").
expand:
Replace "${VAR}" or "$VAR" in the value of path according to the
current environment variables defined in the container (e.g.
"/$VAR/foo").
"""
_args = [
Arg("path", path),
Arg("expand", expand, False),
]
_ctx = self._select("directory", _args)
return Directory(_ctx)
[docs]
async def entrypoint(self) -> list[str]:
"""Retrieves entrypoint to be prepended to the arguments of all commands.
Returns
-------
list[str]
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("entrypoint", _args)
return await _ctx.execute(list[str])
[docs]
async def env_variable(self, name: str) -> str | None:
"""Retrieves the value of the specified environment variable.
Parameters
----------
name:
The name of the environment variable to retrieve (e.g., "PATH").
Returns
-------
str | None
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args = [
Arg("name", name),
]
_ctx = self._select("envVariable", _args)
return await _ctx.execute(str | None)
[docs]
async def env_variables(self) -> list["EnvVariable"]:
"""Retrieves the list of environment variables passed to commands."""
_args: list[Arg] = []
_ctx = self._select("envVariables", _args)
_ctx = EnvVariable(_ctx)._select("id", [])
@dataclass
class Response:
id: EnvVariableID
_ids = await _ctx.execute(list[Response])
return [
EnvVariable(
Client.from_context(_ctx)._select(
"loadEnvVariableFromID",
[Arg("id", v.id)],
)
)
for v in _ids
]
[docs]
async def exit_code(self) -> int:
"""The exit code of the last executed command.
Returns an error if no command was set.
Returns
-------
int
The `Int` scalar type represents non-fractional signed whole
numeric values. Int can represent values between -(2^31) and 2^31
- 1.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("exitCode", _args)
return await _ctx.execute(int)
[docs]
def experimental_with_all_gp_us(self) -> Self:
"""EXPERIMENTAL API! Subject to change/removal at any time.
Configures all available GPUs on the host to be accessible to this
container.
This currently works for Nvidia devices only.
"""
_args: list[Arg] = []
_ctx = self._select("experimentalWithAllGPUs", _args)
return Container(_ctx)
[docs]
def experimental_with_gpu(self, devices: list[str]) -> Self:
"""EXPERIMENTAL API! Subject to change/removal at any time.
Configures the provided list of devices to be accessible to this
container.
This currently works for Nvidia devices only.
Parameters
----------
devices:
List of devices to be accessible to this container.
"""
_args = [
Arg("devices", devices),
]
_ctx = self._select("experimentalWithGPU", _args)
return Container(_ctx)
[docs]
async def export(
self,
path: str,
*,
platform_variants: "list[Container] | None" = None,
forced_compression: ImageLayerCompression | None = None,
media_types: ImageMediaTypes | None = ImageMediaTypes.OCIMediaTypes,
expand: bool | None = False,
) -> str:
"""Writes the container as an OCI tarball to the destination file path on
the host.
It can also export platform variants.
Parameters
----------
path:
Host's destination path (e.g., "./tarball").
Path can be relative to the engine's workdir or absolute.
platform_variants:
Identifiers for other platform specific containers.
Used for multi-platform image.
forced_compression:
Force each layer of the exported image to use the specified
compression algorithm.
If this is unset, then if a layer already has a compressed blob in
the engine's cache, that will be used (this can result in a mix of
compression algorithms for different layers). If this is unset and
a layer has no compressed blob in the engine's cache, then it will
be compressed using Gzip.
media_types:
Use the specified media types for the exported image's layers.
Defaults to OCI, which is largely compatible with most recent
container runtimes, but Docker may be needed for older runtimes
without OCI support.
expand:
Replace "${VAR}" or "$VAR" in the value of path according to the
current environment variables defined in the container (e.g.
"/$VAR/foo").
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args = [
Arg("path", path),
Arg(
"platformVariants",
() if platform_variants is None else platform_variants,
(),
),
Arg("forcedCompression", forced_compression, None),
Arg("mediaTypes", media_types, ImageMediaTypes.OCIMediaTypes),
Arg("expand", expand, False),
]
_ctx = self._select("export", _args)
return await _ctx.execute(str)
[docs]
async def exposed_ports(self) -> list["Port"]:
"""Retrieves the list of exposed ports.
This includes ports already exposed by the image, even if not
explicitly added with dagger.
"""
_args: list[Arg] = []
_ctx = self._select("exposedPorts", _args)
_ctx = Port(_ctx)._select("id", [])
@dataclass
class Response:
id: PortID
_ids = await _ctx.execute(list[Response])
return [
Port(
Client.from_context(_ctx)._select(
"loadPortFromID",
[Arg("id", v.id)],
)
)
for v in _ids
]
[docs]
def file(
self,
path: str,
*,
expand: bool | None = False,
) -> "File":
"""Retrieves a file at the given path.
Mounts are included.
Parameters
----------
path:
The path of the file to retrieve (e.g., "./README.md").
expand:
Replace "${VAR}" or "$VAR" in the value of path according to the
current environment variables defined in the container (e.g.
"/$VAR/foo.txt").
"""
_args = [
Arg("path", path),
Arg("expand", expand, False),
]
_ctx = self._select("file", _args)
return File(_ctx)
[docs]
def from_(self, address: str) -> Self:
"""Initializes this container from a pulled base image.
Parameters
----------
address:
Image's address from its registry.
Formatted as [host]/[user]/[repo]:[tag] (e.g.,
"docker.io/dagger/dagger:main").
"""
_args = [
Arg("address", address),
]
_ctx = self._select("from", _args)
return Container(_ctx)
[docs]
async def id(self) -> ContainerID:
"""A unique identifier for this Container.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
ContainerID
The `ContainerID` scalar type represents an identifier for an
object of type Container.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(ContainerID)
[docs]
async def image_ref(self) -> str:
"""The unique image reference which can only be retrieved immediately
after the 'Container.From' call.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("imageRef", _args)
return await _ctx.execute(str)
[docs]
def import_(
self,
source: "File",
*,
tag: str | None = "",
) -> Self:
"""Reads the container from an OCI tarball.
Parameters
----------
source:
File to read the container from.
tag:
Identifies the tag to import from the archive, if the archive
bundles multiple tags.
"""
_args = [
Arg("source", source),
Arg("tag", tag, ""),
]
_ctx = self._select("import", _args)
return Container(_ctx)
[docs]
async def label(self, name: str) -> str | None:
"""Retrieves the value of the specified label.
Parameters
----------
name:
The name of the label (e.g.,
"org.opencontainers.artifact.created").
Returns
-------
str | None
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args = [
Arg("name", name),
]
_ctx = self._select("label", _args)
return await _ctx.execute(str | None)
[docs]
async def labels(self) -> list["Label"]:
"""Retrieves the list of labels passed to container."""
_args: list[Arg] = []
_ctx = self._select("labels", _args)
_ctx = Label(_ctx)._select("id", [])
@dataclass
class Response:
id: LabelID
_ids = await _ctx.execute(list[Response])
return [
Label(
Client.from_context(_ctx)._select(
"loadLabelFromID",
[Arg("id", v.id)],
)
)
for v in _ids
]
[docs]
async def mounts(self) -> list[str]:
"""Retrieves the list of paths where a directory is mounted.
Returns
-------
list[str]
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("mounts", _args)
return await _ctx.execute(list[str])
[docs]
async def publish(
self,
address: str,
*,
platform_variants: "list[Container] | None" = None,
forced_compression: ImageLayerCompression | None = None,
media_types: ImageMediaTypes | None = ImageMediaTypes.OCIMediaTypes,
) -> str:
"""Publishes this container as a new image to the specified address.
Publish returns a fully qualified ref.
It can also publish platform variants.
Parameters
----------
address:
Registry's address to publish the image to.
Formatted as [host]/[user]/[repo]:[tag] (e.g.
"docker.io/dagger/dagger:main").
platform_variants:
Identifiers for other platform specific containers.
Used for multi-platform image.
forced_compression:
Force each layer of the published image to use the specified
compression algorithm.
If this is unset, then if a layer already has a compressed blob in
the engine's cache, that will be used (this can result in a mix of
compression algorithms for different layers). If this is unset and
a layer has no compressed blob in the engine's cache, then it will
be compressed using Gzip.
media_types:
Use the specified media types for the published image's layers.
Defaults to OCI, which is largely compatible with most recent
registries, but Docker may be needed for older registries without
OCI support.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args = [
Arg("address", address),
Arg(
"platformVariants",
() if platform_variants is None else platform_variants,
(),
),
Arg("forcedCompression", forced_compression, None),
Arg("mediaTypes", media_types, ImageMediaTypes.OCIMediaTypes),
]
_ctx = self._select("publish", _args)
return await _ctx.execute(str)
[docs]
def rootfs(self) -> "Directory":
"""Retrieves this container's root filesystem. Mounts are not included."""
_args: list[Arg] = []
_ctx = self._select("rootfs", _args)
return Directory(_ctx)
[docs]
async def stderr(self) -> str:
"""The error stream of the last executed command.
Returns an error if no command was set.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("stderr", _args)
return await _ctx.execute(str)
[docs]
async def stdout(self) -> str:
"""The output stream of the last executed command.
Returns an error if no command was set.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("stdout", _args)
return await _ctx.execute(str)
[docs]
async def sync(self) -> Self:
"""Forces evaluation of the pipeline in the engine.
It doesn't run the default command if no exec has been set.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("sync", _args)
_id = await _ctx.execute(ContainerID)
_ctx = Client.from_context(_ctx)._select(
"loadContainerFromID", [Arg("id", _id)]
)
return Container(_ctx)
def __await__(self):
return self.sync().__await__()
[docs]
def terminal(
self,
*,
cmd: list[str] | None = None,
experimental_privileged_nesting: bool | None = False,
insecure_root_capabilities: bool | None = False,
) -> Self:
"""Opens an interactive terminal for this container using its configured
default terminal command if not overridden by args (or sh as a
fallback default).
Parameters
----------
cmd:
If set, override the container's default terminal command and
invoke these command arguments instead.
experimental_privileged_nesting:
Provides Dagger access to the executed command.
Do not use this option unless you trust the command being
executed; the command being executed WILL BE GRANTED FULL ACCESS
TO YOUR HOST FILESYSTEM.
insecure_root_capabilities:
Execute the command with all root capabilities. This is similar to
running a command with "sudo" or executing "docker run" with the "
--privileged" flag. Containerization does not provide any security
guarantees when using this option. It should only be used when
absolutely necessary and only with trusted commands.
"""
_args = [
Arg("cmd", () if cmd is None else cmd, ()),
Arg(
"experimentalPrivilegedNesting", experimental_privileged_nesting, False
),
Arg("insecureRootCapabilities", insecure_root_capabilities, False),
]
_ctx = self._select("terminal", _args)
return Container(_ctx)
[docs]
async def up(
self,
*,
ports: list[PortForward] | None = None,
random: bool | None = False,
args: list[str] | None = None,
use_entrypoint: bool | None = False,
experimental_privileged_nesting: bool | None = False,
insecure_root_capabilities: bool | None = False,
expand: bool | None = False,
no_init: bool | None = False,
) -> Void | None:
"""Starts a Service and creates a tunnel that forwards traffic from the
caller's network to that service.
Be sure to set any exposed ports before calling this api.
Parameters
----------
ports:
List of frontend/backend port mappings to forward.
Frontend is the port accepting traffic on the host, backend is the
service port.
random:
Bind each tunnel port to a random port on the host.
args:
Command to run instead of the container's default command (e.g.,
["go", "run", "main.go"]).
If empty, the container's default command is used.
use_entrypoint:
If the container has an entrypoint, prepend it to the args.
experimental_privileged_nesting:
Provides Dagger access to the executed command.
Do not use this option unless you trust the command being
executed; the command being executed WILL BE GRANTED FULL ACCESS
TO YOUR HOST FILESYSTEM.
insecure_root_capabilities:
Execute the command with all root capabilities. This is similar to
running a command with "sudo" or executing "docker run" with the "
--privileged" flag. Containerization does not provide any security
guarantees when using this option. It should only be used when
absolutely necessary and only with trusted commands.
expand:
Replace "${VAR}" or "$VAR" in the args according to the current
environment variables defined in the container (e.g. "/$VAR/foo").
no_init:
If set, skip the automatic init process injected into containers
by default.
This should only be used if the user requires that their exec
process be the pid 1 process in the container. Otherwise it may
result in unexpected behavior.
Returns
-------
Void | None
The absence of a value. A Null Void is used as a placeholder for
resolvers that do not return anything.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args = [
Arg("ports", () if ports is None else ports, ()),
Arg("random", random, False),
Arg("args", () if args is None else args, ()),
Arg("useEntrypoint", use_entrypoint, False),
Arg(
"experimentalPrivilegedNesting", experimental_privileged_nesting, False
),
Arg("insecureRootCapabilities", insecure_root_capabilities, False),
Arg("expand", expand, False),
Arg("noInit", no_init, False),
]
_ctx = self._select("up", _args)
await _ctx.execute()
[docs]
async def user(self) -> str:
"""Retrieves the user to be set for all commands.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("user", _args)
return await _ctx.execute(str)
[docs]
def with_annotation(self, name: str, value: str) -> Self:
"""Retrieves this container plus the given OCI anotation.
Parameters
----------
name:
The name of the annotation.
value:
The value of the annotation.
"""
_args = [
Arg("name", name),
Arg("value", value),
]
_ctx = self._select("withAnnotation", _args)
return Container(_ctx)
[docs]
def with_default_args(self, args: list[str]) -> Self:
"""Configures default arguments for future commands.
Parameters
----------
args:
Arguments to prepend to future executions (e.g., ["-v", "--no-
cache"]).
"""
_args = [
Arg("args", args),
]
_ctx = self._select("withDefaultArgs", _args)
return Container(_ctx)
[docs]
def with_default_terminal_cmd(
self,
args: list[str],
*,
experimental_privileged_nesting: bool | None = False,
insecure_root_capabilities: bool | None = False,
) -> Self:
"""Set the default command to invoke for the container's terminal API.
Parameters
----------
args:
The args of the command.
experimental_privileged_nesting:
Provides Dagger access to the executed command.
Do not use this option unless you trust the command being
executed; the command being executed WILL BE GRANTED FULL ACCESS
TO YOUR HOST FILESYSTEM.
insecure_root_capabilities:
Execute the command with all root capabilities. This is similar to
running a command with "sudo" or executing "docker run" with the "
--privileged" flag. Containerization does not provide any security
guarantees when using this option. It should only be used when
absolutely necessary and only with trusted commands.
"""
_args = [
Arg("args", args),
Arg(
"experimentalPrivilegedNesting", experimental_privileged_nesting, False
),
Arg("insecureRootCapabilities", insecure_root_capabilities, False),
]
_ctx = self._select("withDefaultTerminalCmd", _args)
return Container(_ctx)
[docs]
def with_directory(
self,
path: str,
directory: "Directory",
*,
exclude: list[str] | None = None,
include: list[str] | None = None,
owner: str | None = "",
expand: bool | None = False,
) -> Self:
"""Retrieves this container plus a directory written at the given path.
Parameters
----------
path:
Location of the written directory (e.g., "/tmp/directory").
directory:
Identifier of the directory to write
exclude:
Patterns to exclude in the written directory (e.g.
["node_modules/**", ".gitignore", ".git/"]).
include:
Patterns to include in the written directory (e.g. ["*.go",
"go.mod", "go.sum"]).
owner:
A user:group to set for the directory and its contents.
The user and group can either be an ID (1000:1000) or a name
(foo:bar).
If the group is omitted, it defaults to the same as the user.
expand:
Replace "${VAR}" or "$VAR" in the value of path according to the
current environment variables defined in the container (e.g.
"/$VAR/foo").
"""
_args = [
Arg("path", path),
Arg("directory", directory),
Arg("exclude", () if exclude is None else exclude, ()),
Arg("include", () if include is None else include, ()),
Arg("owner", owner, ""),
Arg("expand", expand, False),
]
_ctx = self._select("withDirectory", _args)
return Container(_ctx)
[docs]
def with_entrypoint(
self,
args: list[str],
*,
keep_default_args: bool | None = False,
) -> Self:
"""Retrieves this container but with a different command entrypoint.
Parameters
----------
args:
Entrypoint to use for future executions (e.g., ["go", "run"]).
keep_default_args:
Don't remove the default arguments when setting the entrypoint.
"""
_args = [
Arg("args", args),
Arg("keepDefaultArgs", keep_default_args, False),
]
_ctx = self._select("withEntrypoint", _args)
return Container(_ctx)
[docs]
def with_env_variable(
self,
name: str,
value: str,
*,
expand: bool | None = False,
) -> Self:
"""Retrieves this container plus the given environment variable.
Parameters
----------
name:
The name of the environment variable (e.g., "HOST").
value:
The value of the environment variable. (e.g., "localhost").
expand:
Replace "${VAR}" or "$VAR" in the value according to the current
environment variables defined in the container (e.g.
"/opt/bin:$PATH").
"""
_args = [
Arg("name", name),
Arg("value", value),
Arg("expand", expand, False),
]
_ctx = self._select("withEnvVariable", _args)
return Container(_ctx)
[docs]
def with_exec(
self,
args: list[str],
*,
use_entrypoint: bool | None = False,
stdin: str | None = "",
redirect_stdout: str | None = "",
redirect_stderr: str | None = "",
expect: ReturnType | None = ReturnType.SUCCESS,
experimental_privileged_nesting: bool | None = False,
insecure_root_capabilities: bool | None = False,
expand: bool | None = False,
no_init: bool | None = False,
) -> Self:
"""Retrieves this container after executing the specified command inside
it.
Parameters
----------
args:
Command to run instead of the container's default command (e.g.,
["go", "run", "main.go"]).
If empty, the container's default command is used.
use_entrypoint:
If the container has an entrypoint, prepend it to the args.
stdin:
Content to write to the command's standard input before closing
(e.g., "Hello world").
redirect_stdout:
Redirect the command's standard output to a file in the container
(e.g., "/tmp/stdout").
redirect_stderr:
Redirect the command's standard error to a file in the container
(e.g., "/tmp/stderr").
expect:
Exit codes this command is allowed to exit with without error
experimental_privileged_nesting:
Provides Dagger access to the executed command.
Do not use this option unless you trust the command being
executed; the command being executed WILL BE GRANTED FULL ACCESS
TO YOUR HOST FILESYSTEM.
insecure_root_capabilities:
Execute the command with all root capabilities. This is similar to
running a command with "sudo" or executing "docker run" with the "
--privileged" flag. Containerization does not provide any security
guarantees when using this option. It should only be used when
absolutely necessary and only with trusted commands.
expand:
Replace "${VAR}" or "$VAR" in the args according to the current
environment variables defined in the container (e.g. "/$VAR/foo").
no_init:
If set, skip the automatic init process injected into containers
by default.
This should only be used if the user requires that their exec
process be the pid 1 process in the container. Otherwise it may
result in unexpected behavior.
"""
_args = [
Arg("args", args),
Arg("useEntrypoint", use_entrypoint, False),
Arg("stdin", stdin, ""),
Arg("redirectStdout", redirect_stdout, ""),
Arg("redirectStderr", redirect_stderr, ""),
Arg("expect", expect, ReturnType.SUCCESS),
Arg(
"experimentalPrivilegedNesting", experimental_privileged_nesting, False
),
Arg("insecureRootCapabilities", insecure_root_capabilities, False),
Arg("expand", expand, False),
Arg("noInit", no_init, False),
]
_ctx = self._select("withExec", _args)
return Container(_ctx)
[docs]
def with_exposed_port(
self,
port: int,
*,
protocol: NetworkProtocol | None = NetworkProtocol.TCP,
description: str | None = None,
experimental_skip_healthcheck: bool | None = False,
) -> Self:
"""Expose a network port.
Exposed ports serve two purposes:
- For health checks and introspection, when running services
- For setting the EXPOSE OCI field when publishing the container
Parameters
----------
port:
Port number to expose
protocol:
Transport layer network protocol
description:
Optional port description
experimental_skip_healthcheck:
Skip the health check when run as a service.
"""
_args = [
Arg("port", port),
Arg("protocol", protocol, NetworkProtocol.TCP),
Arg("description", description, None),
Arg("experimentalSkipHealthcheck", experimental_skip_healthcheck, False),
]
_ctx = self._select("withExposedPort", _args)
return Container(_ctx)
[docs]
def with_file(
self,
path: str,
source: "File",
*,
permissions: int | None = None,
owner: str | None = "",
expand: bool | None = False,
) -> Self:
"""Retrieves this container plus the contents of the given file copied to
the given path.
Parameters
----------
path:
Location of the copied file (e.g., "/tmp/file.txt").
source:
Identifier of the file to copy.
permissions:
Permission given to the copied file (e.g., 0600).
owner:
A user:group to set for the file.
The user and group can either be an ID (1000:1000) or a name
(foo:bar).
If the group is omitted, it defaults to the same as the user.
expand:
Replace "${VAR}" or "$VAR" in the value of path according to the
current environment variables defined in the container (e.g.
"/$VAR/foo.txt").
"""
_args = [
Arg("path", path),
Arg("source", source),
Arg("permissions", permissions, None),
Arg("owner", owner, ""),
Arg("expand", expand, False),
]
_ctx = self._select("withFile", _args)
return Container(_ctx)
[docs]
def with_files(
self,
path: str,
sources: list["File"],
*,
permissions: int | None = None,
owner: str | None = "",
expand: bool | None = False,
) -> Self:
"""Retrieves this container plus the contents of the given files copied
to the given path.
Parameters
----------
path:
Location where copied files should be placed (e.g., "/src").
sources:
Identifiers of the files to copy.
permissions:
Permission given to the copied files (e.g., 0600).
owner:
A user:group to set for the files.
The user and group can either be an ID (1000:1000) or a name
(foo:bar).
If the group is omitted, it defaults to the same as the user.
expand:
Replace "${VAR}" or "$VAR" in the value of path according to the
current environment variables defined in the container (e.g.
"/$VAR/foo.txt").
"""
_args = [
Arg("path", path),
Arg("sources", sources),
Arg("permissions", permissions, None),
Arg("owner", owner, ""),
Arg("expand", expand, False),
]
_ctx = self._select("withFiles", _args)
return Container(_ctx)
[docs]
def with_label(self, name: str, value: str) -> Self:
"""Retrieves this container plus the given label.
Parameters
----------
name:
The name of the label (e.g.,
"org.opencontainers.artifact.created").
value:
The value of the label (e.g., "2023-01-01T00:00:00Z").
"""
_args = [
Arg("name", name),
Arg("value", value),
]
_ctx = self._select("withLabel", _args)
return Container(_ctx)
[docs]
def with_mounted_cache(
self,
path: str,
cache: CacheVolume,
*,
source: "Directory | None" = None,
sharing: CacheSharingMode | None = CacheSharingMode.SHARED,
owner: str | None = "",
expand: bool | None = False,
) -> Self:
"""Retrieves this container plus a cache volume mounted at the given
path.
Parameters
----------
path:
Location of the cache directory (e.g., "/root/.npm").
cache:
Identifier of the cache volume to mount.
source:
Identifier of the directory to use as the cache volume's root.
sharing:
Sharing mode of the cache volume.
owner:
A user:group to set for the mounted cache directory.
Note that this changes the ownership of the specified mount along
with the initial filesystem provided by source (if any). It does
not have any effect if/when the cache has already been created.
The user and group can either be an ID (1000:1000) or a name
(foo:bar).
If the group is omitted, it defaults to the same as the user.
expand:
Replace "${VAR}" or "$VAR" in the value of path according to the
current environment variables defined in the container (e.g.
"/$VAR/foo").
"""
_args = [
Arg("path", path),
Arg("cache", cache),
Arg("source", source, None),
Arg("sharing", sharing, CacheSharingMode.SHARED),
Arg("owner", owner, ""),
Arg("expand", expand, False),
]
_ctx = self._select("withMountedCache", _args)
return Container(_ctx)
[docs]
def with_mounted_directory(
self,
path: str,
source: "Directory",
*,
owner: str | None = "",
expand: bool | None = False,
) -> Self:
"""Retrieves this container plus a directory mounted at the given path.
Parameters
----------
path:
Location of the mounted directory (e.g., "/mnt/directory").
source:
Identifier of the mounted directory.
owner:
A user:group to set for the mounted directory and its contents.
The user and group can either be an ID (1000:1000) or a name
(foo:bar).
If the group is omitted, it defaults to the same as the user.
expand:
Replace "${VAR}" or "$VAR" in the value of path according to the
current environment variables defined in the container (e.g.
"/$VAR/foo").
"""
_args = [
Arg("path", path),
Arg("source", source),
Arg("owner", owner, ""),
Arg("expand", expand, False),
]
_ctx = self._select("withMountedDirectory", _args)
return Container(_ctx)
[docs]
def with_mounted_file(
self,
path: str,
source: "File",
*,
owner: str | None = "",
expand: bool | None = False,
) -> Self:
"""Retrieves this container plus a file mounted at the given path.
Parameters
----------
path:
Location of the mounted file (e.g., "/tmp/file.txt").
source:
Identifier of the mounted file.
owner:
A user or user:group to set for the mounted file.
The user and group can either be an ID (1000:1000) or a name
(foo:bar).
If the group is omitted, it defaults to the same as the user.
expand:
Replace "${VAR}" or "$VAR" in the value of path according to the
current environment variables defined in the container (e.g.
"/$VAR/foo.txt").
"""
_args = [
Arg("path", path),
Arg("source", source),
Arg("owner", owner, ""),
Arg("expand", expand, False),
]
_ctx = self._select("withMountedFile", _args)
return Container(_ctx)
[docs]
def with_mounted_secret(
self,
path: str,
source: "Secret",
*,
owner: str | None = "",
mode: int | None = 256,
expand: bool | None = False,
) -> Self:
"""Retrieves this container plus a secret mounted into a file at the
given path.
Parameters
----------
path:
Location of the secret file (e.g., "/tmp/secret.txt").
source:
Identifier of the secret to mount.
owner:
A user:group to set for the mounted secret.
The user and group can either be an ID (1000:1000) or a name
(foo:bar).
If the group is omitted, it defaults to the same as the user.
mode:
Permission given to the mounted secret (e.g., 0600).
This option requires an owner to be set to be active.
expand:
Replace "${VAR}" or "$VAR" in the value of path according to the
current environment variables defined in the container (e.g.
"/$VAR/foo").
"""
_args = [
Arg("path", path),
Arg("source", source),
Arg("owner", owner, ""),
Arg("mode", mode, 256),
Arg("expand", expand, False),
]
_ctx = self._select("withMountedSecret", _args)
return Container(_ctx)
[docs]
def with_mounted_temp(
self,
path: str,
*,
size: int | None = None,
expand: bool | None = False,
) -> Self:
"""Retrieves this container plus a temporary directory mounted at the
given path. Any writes will be ephemeral to a single withExec call;
they will not be persisted to subsequent withExecs.
Parameters
----------
path:
Location of the temporary directory (e.g., "/tmp/temp_dir").
size:
Size of the temporary directory in bytes.
expand:
Replace "${VAR}" or "$VAR" in the value of path according to the
current environment variables defined in the container (e.g.
"/$VAR/foo").
"""
_args = [
Arg("path", path),
Arg("size", size, None),
Arg("expand", expand, False),
]
_ctx = self._select("withMountedTemp", _args)
return Container(_ctx)
[docs]
def with_new_file(
self,
path: str,
contents: str,
*,
permissions: int | None = 420,
owner: str | None = "",
expand: bool | None = False,
) -> Self:
"""Retrieves this container plus a new file written at the given path.
Parameters
----------
path:
Location of the written file (e.g., "/tmp/file.txt").
contents:
Content of the file to write (e.g., "Hello world!").
permissions:
Permission given to the written file (e.g., 0600).
owner:
A user:group to set for the file.
The user and group can either be an ID (1000:1000) or a name
(foo:bar).
If the group is omitted, it defaults to the same as the user.
expand:
Replace "${VAR}" or "$VAR" in the value of path according to the
current environment variables defined in the container (e.g.
"/$VAR/foo.txt").
"""
_args = [
Arg("path", path),
Arg("contents", contents),
Arg("permissions", permissions, 420),
Arg("owner", owner, ""),
Arg("expand", expand, False),
]
_ctx = self._select("withNewFile", _args)
return Container(_ctx)
[docs]
def with_registry_auth(
self,
address: str,
username: str,
secret: "Secret",
) -> Self:
"""Retrieves this container with a registry authentication for a given
address.
Parameters
----------
address:
Registry's address to bind the authentication to.
Formatted as [host]/[user]/[repo]:[tag] (e.g.
docker.io/dagger/dagger:main).
username:
The username of the registry's account (e.g., "Dagger").
secret:
The API key, password or token to authenticate to this registry.
"""
_args = [
Arg("address", address),
Arg("username", username),
Arg("secret", secret),
]
_ctx = self._select("withRegistryAuth", _args)
return Container(_ctx)
[docs]
def with_rootfs(self, directory: "Directory") -> Self:
"""Retrieves the container with the given directory mounted to /.
Parameters
----------
directory:
Directory to mount.
"""
_args = [
Arg("directory", directory),
]
_ctx = self._select("withRootfs", _args)
return Container(_ctx)
[docs]
def with_secret_variable(self, name: str, secret: "Secret") -> Self:
"""Retrieves this container plus an env variable containing the given
secret.
Parameters
----------
name:
The name of the secret variable (e.g., "API_SECRET").
secret:
The identifier of the secret value.
"""
_args = [
Arg("name", name),
Arg("secret", secret),
]
_ctx = self._select("withSecretVariable", _args)
return Container(_ctx)
[docs]
def with_service_binding(self, alias: str, service: "Service") -> Self:
"""Establish a runtime dependency on a service.
The service will be started automatically when needed and detached
when it is no longer needed, executing the default command if none is
set.
The service will be reachable from the container via the provided
hostname alias.
The service dependency will also convey to any files or directories
produced by the container.
Parameters
----------
alias:
A name that can be used to reach the service from the container
service:
Identifier of the service container
"""
_args = [
Arg("alias", alias),
Arg("service", service),
]
_ctx = self._select("withServiceBinding", _args)
return Container(_ctx)
[docs]
def with_unix_socket(
self,
path: str,
source: "Socket",
*,
owner: str | None = "",
expand: bool | None = False,
) -> Self:
"""Retrieves this container plus a socket forwarded to the given Unix
socket path.
Parameters
----------
path:
Location of the forwarded Unix socket (e.g., "/tmp/socket").
source:
Identifier of the socket to forward.
owner:
A user:group to set for the mounted socket.
The user and group can either be an ID (1000:1000) or a name
(foo:bar).
If the group is omitted, it defaults to the same as the user.
expand:
Replace "${VAR}" or "$VAR" in the value of path according to the
current environment variables defined in the container (e.g.
"/$VAR/foo").
"""
_args = [
Arg("path", path),
Arg("source", source),
Arg("owner", owner, ""),
Arg("expand", expand, False),
]
_ctx = self._select("withUnixSocket", _args)
return Container(_ctx)
[docs]
def with_user(self, name: str) -> Self:
"""Retrieves this container with a different command user.
Parameters
----------
name:
The user to set (e.g., "root").
"""
_args = [
Arg("name", name),
]
_ctx = self._select("withUser", _args)
return Container(_ctx)
[docs]
def with_workdir(
self,
path: str,
*,
expand: bool | None = False,
) -> Self:
"""Retrieves this container with a different working directory.
Parameters
----------
path:
The path to set as the working directory (e.g., "/app").
expand:
Replace "${VAR}" or "$VAR" in the value of path according to the
current environment variables defined in the container (e.g.
"/$VAR/foo").
"""
_args = [
Arg("path", path),
Arg("expand", expand, False),
]
_ctx = self._select("withWorkdir", _args)
return Container(_ctx)
[docs]
def without_annotation(self, name: str) -> Self:
"""Retrieves this container minus the given OCI annotation.
Parameters
----------
name:
The name of the annotation.
"""
_args = [
Arg("name", name),
]
_ctx = self._select("withoutAnnotation", _args)
return Container(_ctx)
[docs]
def without_default_args(self) -> Self:
"""Retrieves this container with unset default arguments for future
commands.
"""
_args: list[Arg] = []
_ctx = self._select("withoutDefaultArgs", _args)
return Container(_ctx)
[docs]
def without_directory(
self,
path: str,
*,
expand: bool | None = False,
) -> Self:
"""Retrieves this container with the directory at the given path removed.
Parameters
----------
path:
Location of the directory to remove (e.g., ".github/").
expand:
Replace "${VAR}" or "$VAR" in the value of path according to the
current environment variables defined in the container (e.g.
"/$VAR/foo").
"""
_args = [
Arg("path", path),
Arg("expand", expand, False),
]
_ctx = self._select("withoutDirectory", _args)
return Container(_ctx)
[docs]
def without_entrypoint(
self,
*,
keep_default_args: bool | None = False,
) -> Self:
"""Retrieves this container with an unset command entrypoint.
Parameters
----------
keep_default_args:
Don't remove the default arguments when unsetting the entrypoint.
"""
_args = [
Arg("keepDefaultArgs", keep_default_args, False),
]
_ctx = self._select("withoutEntrypoint", _args)
return Container(_ctx)
[docs]
def without_env_variable(self, name: str) -> Self:
"""Retrieves this container minus the given environment variable.
Parameters
----------
name:
The name of the environment variable (e.g., "HOST").
"""
_args = [
Arg("name", name),
]
_ctx = self._select("withoutEnvVariable", _args)
return Container(_ctx)
[docs]
def without_exposed_port(
self,
port: int,
*,
protocol: NetworkProtocol | None = NetworkProtocol.TCP,
) -> Self:
"""Unexpose a previously exposed port.
Parameters
----------
port:
Port number to unexpose
protocol:
Port protocol to unexpose
"""
_args = [
Arg("port", port),
Arg("protocol", protocol, NetworkProtocol.TCP),
]
_ctx = self._select("withoutExposedPort", _args)
return Container(_ctx)
[docs]
def without_file(
self,
path: str,
*,
expand: bool | None = False,
) -> Self:
"""Retrieves this container with the file at the given path removed.
Parameters
----------
path:
Location of the file to remove (e.g., "/file.txt").
expand:
Replace "${VAR}" or "$VAR" in the value of path according to the
current environment variables defined in the container (e.g.
"/$VAR/foo.txt").
"""
_args = [
Arg("path", path),
Arg("expand", expand, False),
]
_ctx = self._select("withoutFile", _args)
return Container(_ctx)
[docs]
def without_files(
self,
paths: list[str],
*,
expand: bool | None = False,
) -> Self:
"""Retrieves this container with the files at the given paths removed.
Parameters
----------
paths:
Location of the files to remove (e.g., ["/file.txt"]).
expand:
Replace "${VAR}" or "$VAR" in the value of paths according to the
current environment variables defined in the container (e.g.
"/$VAR/foo.txt").
"""
_args = [
Arg("paths", paths),
Arg("expand", expand, False),
]
_ctx = self._select("withoutFiles", _args)
return Container(_ctx)
[docs]
def without_label(self, name: str) -> Self:
"""Retrieves this container minus the given environment label.
Parameters
----------
name:
The name of the label to remove (e.g.,
"org.opencontainers.artifact.created").
"""
_args = [
Arg("name", name),
]
_ctx = self._select("withoutLabel", _args)
return Container(_ctx)
[docs]
def without_mount(
self,
path: str,
*,
expand: bool | None = False,
) -> Self:
"""Retrieves this container after unmounting everything at the given
path.
Parameters
----------
path:
Location of the cache directory (e.g., "/root/.npm").
expand:
Replace "${VAR}" or "$VAR" in the value of path according to the
current environment variables defined in the container (e.g.
"/$VAR/foo").
"""
_args = [
Arg("path", path),
Arg("expand", expand, False),
]
_ctx = self._select("withoutMount", _args)
return Container(_ctx)
[docs]
def without_registry_auth(self, address: str) -> Self:
"""Retrieves this container without the registry authentication of a
given address.
Parameters
----------
address:
Registry's address to remove the authentication from.
Formatted as [host]/[user]/[repo]:[tag] (e.g.
docker.io/dagger/dagger:main).
"""
_args = [
Arg("address", address),
]
_ctx = self._select("withoutRegistryAuth", _args)
return Container(_ctx)
[docs]
def without_secret_variable(self, name: str) -> Self:
"""Retrieves this container minus the given environment variable
containing the secret.
Parameters
----------
name:
The name of the environment variable (e.g., "HOST").
"""
_args = [
Arg("name", name),
]
_ctx = self._select("withoutSecretVariable", _args)
return Container(_ctx)
[docs]
def without_unix_socket(
self,
path: str,
*,
expand: bool | None = False,
) -> Self:
"""Retrieves this container with a previously added Unix socket removed.
Parameters
----------
path:
Location of the socket to remove (e.g., "/tmp/socket").
expand:
Replace "${VAR}" or "$VAR" in the value of path according to the
current environment variables defined in the container (e.g.
"/$VAR/foo").
"""
_args = [
Arg("path", path),
Arg("expand", expand, False),
]
_ctx = self._select("withoutUnixSocket", _args)
return Container(_ctx)
[docs]
def without_user(self) -> Self:
"""Retrieves this container with an unset command user.
Should default to root.
"""
_args: list[Arg] = []
_ctx = self._select("withoutUser", _args)
return Container(_ctx)
[docs]
def without_workdir(self) -> Self:
"""Retrieves this container with an unset working directory.
Should default to "/".
"""
_args: list[Arg] = []
_ctx = self._select("withoutWorkdir", _args)
return Container(_ctx)
[docs]
async def workdir(self) -> str:
"""Retrieves the working directory for all commands.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("workdir", _args)
return await _ctx.execute(str)
[docs]
def with_(self, cb: Callable[["Container"], "Container"]) -> "Container":
"""Call the provided callable with current Container.
This is useful for reusability and readability by not breaking the calling chain.
"""
return cb(self)
@typecheck
class CurrentModule(Type):
"""Reflective module API provided to functions at runtime."""
[docs]
async def id(self) -> CurrentModuleID:
"""A unique identifier for this CurrentModule.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
CurrentModuleID
The `CurrentModuleID` scalar type represents an identifier for an
object of type CurrentModule.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(CurrentModuleID)
[docs]
async def name(self) -> str:
"""The name of the module being executed in
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("name", _args)
return await _ctx.execute(str)
[docs]
def source(self) -> "Directory":
"""The directory containing the module's source code loaded into the
engine (plus any generated code that may have been created).
"""
_args: list[Arg] = []
_ctx = self._select("source", _args)
return Directory(_ctx)
[docs]
def workdir(
self,
path: str,
*,
exclude: list[str] | None = None,
include: list[str] | None = None,
) -> "Directory":
"""Load a directory from the module's scratch working directory,
including any changes that may have been made to it during module
function execution.
Parameters
----------
path:
Location of the directory to access (e.g., ".").
exclude:
Exclude artifacts that match the given pattern (e.g.,
["node_modules/", ".git*"]).
include:
Include only artifacts that match the given pattern (e.g.,
["app/", "package.*"]).
"""
_args = [
Arg("path", path),
Arg("exclude", () if exclude is None else exclude, ()),
Arg("include", () if include is None else include, ()),
]
_ctx = self._select("workdir", _args)
return Directory(_ctx)
[docs]
def workdir_file(self, path: str) -> "File":
"""Load a file from the module's scratch working directory, including any
changes that may have been made to it during module function
execution.Load a file from the module's scratch working directory,
including any changes that may have been made to it during module
function execution.
Parameters
----------
path:
Location of the file to retrieve (e.g., "README.md").
"""
_args = [
Arg("path", path),
]
_ctx = self._select("workdirFile", _args)
return File(_ctx)
@typecheck
class Directory(Type):
"""A directory."""
[docs]
def as_module(
self,
*,
source_root_path: str | None = ".",
engine_version: str | None = None,
) -> "Module":
"""Load the directory as a Dagger module
Parameters
----------
source_root_path:
An optional subpath of the directory which contains the module's
configuration file.
This is needed when the module code is in a subdirectory but
requires parent directories to be loaded in order to execute. For
example, the module source code may need a go.mod, project.toml,
package.json, etc. file from a parent directory.
If not set, the module source code is loaded from the root of the
directory.
engine_version:
The engine version to upgrade to.
"""
_args = [
Arg("sourceRootPath", source_root_path, "."),
Arg("engineVersion", engine_version, None),
]
_ctx = self._select("asModule", _args)
return Module(_ctx)
[docs]
def diff(self, other: Self) -> Self:
"""Gets the difference between this directory and an another directory.
Parameters
----------
other:
Identifier of the directory to compare.
"""
_args = [
Arg("other", other),
]
_ctx = self._select("diff", _args)
return Directory(_ctx)
[docs]
async def digest(self) -> str:
"""Return the directory's digest. The format of the digest is not
guaranteed to be stable between releases of Dagger. It is guaranteed
to be stable between invocations of the same Dagger engine.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("digest", _args)
return await _ctx.execute(str)
[docs]
def directory(self, path: str) -> Self:
"""Retrieves a directory at the given path.
Parameters
----------
path:
Location of the directory to retrieve (e.g., "/src").
"""
_args = [
Arg("path", path),
]
_ctx = self._select("directory", _args)
return Directory(_ctx)
[docs]
def docker_build(
self,
*,
platform: Platform | None = None,
dockerfile: str | None = "Dockerfile",
target: str | None = "",
build_args: list[BuildArg] | None = None,
secrets: "list[Secret] | None" = None,
) -> Container:
"""Builds a new Docker container from this directory.
Parameters
----------
platform:
The platform to build.
dockerfile:
Path to the Dockerfile to use (e.g., "frontend.Dockerfile").
target:
Target build stage to build.
build_args:
Build arguments to use in the build.
secrets:
Secrets to pass to the build.
They will be mounted at /run/secrets/[secret-name].
"""
_args = [
Arg("platform", platform, None),
Arg("dockerfile", dockerfile, "Dockerfile"),
Arg("target", target, ""),
Arg("buildArgs", () if build_args is None else build_args, ()),
Arg("secrets", () if secrets is None else secrets, ()),
]
_ctx = self._select("dockerBuild", _args)
return Container(_ctx)
[docs]
async def entries(self, *, path: str | None = None) -> list[str]:
"""Returns a list of files and directories at the given path.
Parameters
----------
path:
Location of the directory to look at (e.g., "/src").
Returns
-------
list[str]
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args = [
Arg("path", path, None),
]
_ctx = self._select("entries", _args)
return await _ctx.execute(list[str])
[docs]
async def export(
self,
path: str,
*,
wipe: bool | None = False,
) -> str:
"""Writes the contents of the directory to a path on the host.
Parameters
----------
path:
Location of the copied directory (e.g., "logs/").
wipe:
If true, then the host directory will be wiped clean before
exporting so that it exactly matches the directory being exported;
this means it will delete any files on the host that aren't in the
exported dir. If false (the default), the contents of the
directory will be merged with any existing contents of the host
directory, leaving any existing files on the host that aren't in
the exported directory alone.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args = [
Arg("path", path),
Arg("wipe", wipe, False),
]
_ctx = self._select("export", _args)
return await _ctx.execute(str)
[docs]
def file(self, path: str) -> "File":
"""Retrieves a file at the given path.
Parameters
----------
path:
Location of the file to retrieve (e.g., "README.md").
"""
_args = [
Arg("path", path),
]
_ctx = self._select("file", _args)
return File(_ctx)
[docs]
async def glob(self, pattern: str) -> list[str]:
"""Returns a list of files and directories that matche the given pattern.
Parameters
----------
pattern:
Pattern to match (e.g., "*.md").
Returns
-------
list[str]
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args = [
Arg("pattern", pattern),
]
_ctx = self._select("glob", _args)
return await _ctx.execute(list[str])
[docs]
async def id(self) -> DirectoryID:
"""A unique identifier for this Directory.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
DirectoryID
The `DirectoryID` scalar type represents an identifier for an
object of type Directory.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(DirectoryID)
[docs]
async def sync(self) -> Self:
"""Force evaluation in the engine.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("sync", _args)
_id = await _ctx.execute(DirectoryID)
_ctx = Client.from_context(_ctx)._select(
"loadDirectoryFromID", [Arg("id", _id)]
)
return Directory(_ctx)
def __await__(self):
return self.sync().__await__()
[docs]
def terminal(
self,
*,
cmd: list[str] | None = None,
experimental_privileged_nesting: bool | None = False,
insecure_root_capabilities: bool | None = False,
container: Container | None = None,
) -> Self:
"""Opens an interactive terminal in new container with this directory
mounted inside.
Parameters
----------
cmd:
If set, override the container's default terminal command and
invoke these command arguments instead.
experimental_privileged_nesting:
Provides Dagger access to the executed command.
Do not use this option unless you trust the command being
executed; the command being executed WILL BE GRANTED FULL ACCESS
TO YOUR HOST FILESYSTEM.
insecure_root_capabilities:
Execute the command with all root capabilities. This is similar to
running a command with "sudo" or executing "docker run" with the "
--privileged" flag. Containerization does not provide any security
guarantees when using this option. It should only be used when
absolutely necessary and only with trusted commands.
container:
If set, override the default container used for the terminal.
"""
_args = [
Arg("cmd", () if cmd is None else cmd, ()),
Arg(
"experimentalPrivilegedNesting", experimental_privileged_nesting, False
),
Arg("insecureRootCapabilities", insecure_root_capabilities, False),
Arg("container", container, None),
]
_ctx = self._select("terminal", _args)
return Directory(_ctx)
[docs]
def with_directory(
self,
path: str,
directory: Self,
*,
exclude: list[str] | None = None,
include: list[str] | None = None,
) -> Self:
"""Retrieves this directory plus a directory written at the given path.
Parameters
----------
path:
Location of the written directory (e.g., "/src/").
directory:
Identifier of the directory to copy.
exclude:
Exclude artifacts that match the given pattern (e.g.,
["node_modules/", ".git*"]).
include:
Include only artifacts that match the given pattern (e.g.,
["app/", "package.*"]).
"""
_args = [
Arg("path", path),
Arg("directory", directory),
Arg("exclude", () if exclude is None else exclude, ()),
Arg("include", () if include is None else include, ()),
]
_ctx = self._select("withDirectory", _args)
return Directory(_ctx)
[docs]
def with_file(
self,
path: str,
source: "File",
*,
permissions: int | None = None,
) -> Self:
"""Retrieves this directory plus the contents of the given file copied to
the given path.
Parameters
----------
path:
Location of the copied file (e.g., "/file.txt").
source:
Identifier of the file to copy.
permissions:
Permission given to the copied file (e.g., 0600).
"""
_args = [
Arg("path", path),
Arg("source", source),
Arg("permissions", permissions, None),
]
_ctx = self._select("withFile", _args)
return Directory(_ctx)
[docs]
def with_files(
self,
path: str,
sources: list["File"],
*,
permissions: int | None = None,
) -> Self:
"""Retrieves this directory plus the contents of the given files copied
to the given path.
Parameters
----------
path:
Location where copied files should be placed (e.g., "/src").
sources:
Identifiers of the files to copy.
permissions:
Permission given to the copied files (e.g., 0600).
"""
_args = [
Arg("path", path),
Arg("sources", sources),
Arg("permissions", permissions, None),
]
_ctx = self._select("withFiles", _args)
return Directory(_ctx)
[docs]
def with_new_directory(
self,
path: str,
*,
permissions: int | None = 420,
) -> Self:
"""Retrieves this directory plus a new directory created at the given
path.
Parameters
----------
path:
Location of the directory created (e.g., "/logs").
permissions:
Permission granted to the created directory (e.g., 0777).
"""
_args = [
Arg("path", path),
Arg("permissions", permissions, 420),
]
_ctx = self._select("withNewDirectory", _args)
return Directory(_ctx)
[docs]
def with_new_file(
self,
path: str,
contents: str,
*,
permissions: int | None = 420,
) -> Self:
"""Retrieves this directory plus a new file written at the given path.
Parameters
----------
path:
Location of the written file (e.g., "/file.txt").
contents:
Content of the written file (e.g., "Hello world!").
permissions:
Permission given to the copied file (e.g., 0600).
"""
_args = [
Arg("path", path),
Arg("contents", contents),
Arg("permissions", permissions, 420),
]
_ctx = self._select("withNewFile", _args)
return Directory(_ctx)
[docs]
def with_timestamps(self, timestamp: int) -> Self:
"""Retrieves this directory with all file/dir timestamps set to the given
time.
Parameters
----------
timestamp:
Timestamp to set dir/files in.
Formatted in seconds following Unix epoch (e.g., 1672531199).
"""
_args = [
Arg("timestamp", timestamp),
]
_ctx = self._select("withTimestamps", _args)
return Directory(_ctx)
[docs]
def without_directory(self, path: str) -> Self:
"""Retrieves this directory with the directory at the given path removed.
Parameters
----------
path:
Location of the directory to remove (e.g., ".github/").
"""
_args = [
Arg("path", path),
]
_ctx = self._select("withoutDirectory", _args)
return Directory(_ctx)
[docs]
def without_file(self, path: str) -> Self:
"""Retrieves this directory with the file at the given path removed.
Parameters
----------
path:
Location of the file to remove (e.g., "/file.txt").
"""
_args = [
Arg("path", path),
]
_ctx = self._select("withoutFile", _args)
return Directory(_ctx)
[docs]
def without_files(self, paths: list[str]) -> Self:
"""Retrieves this directory with the files at the given paths removed.
Parameters
----------
paths:
Location of the file to remove (e.g., ["/file.txt"]).
"""
_args = [
Arg("paths", paths),
]
_ctx = self._select("withoutFiles", _args)
return Directory(_ctx)
[docs]
def with_(self, cb: Callable[["Directory"], "Directory"]) -> "Directory":
"""Call the provided callable with current Directory.
This is useful for reusability and readability by not breaking the calling chain.
"""
return cb(self)
@typecheck
class Engine(Type):
"""The Dagger engine configuration and state"""
[docs]
async def id(self) -> EngineID:
"""A unique identifier for this Engine.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
EngineID
The `EngineID` scalar type represents an identifier for an object
of type Engine.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(EngineID)
[docs]
def local_cache(self) -> "EngineCache":
"""The local (on-disk) cache for the Dagger engine"""
_args: list[Arg] = []
_ctx = self._select("localCache", _args)
return EngineCache(_ctx)
@typecheck
class EngineCache(Type):
"""A cache storage for the Dagger engine"""
[docs]
def entry_set(self, *, key: str | None = "") -> "EngineCacheEntrySet":
"""The current set of entries in the cache"""
_args = [
Arg("key", key, ""),
]
_ctx = self._select("entrySet", _args)
return EngineCacheEntrySet(_ctx)
[docs]
async def id(self) -> EngineCacheID:
"""A unique identifier for this EngineCache.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
EngineCacheID
The `EngineCacheID` scalar type represents an identifier for an
object of type EngineCache.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(EngineCacheID)
[docs]
async def keep_bytes(self) -> int:
"""The maximum bytes to keep in the cache without pruning, after which
automatic pruning may kick in.
.. deprecated::
Use minFreeSpace instead.
Returns
-------
int
The `Int` scalar type represents non-fractional signed whole
numeric values. Int can represent values between -(2^31) and 2^31
- 1.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
warnings.warn(
'Method "keep_bytes" is deprecated: Use minFreeSpace instead.',
DeprecationWarning,
stacklevel=4,
)
_args: list[Arg] = []
_ctx = self._select("keepBytes", _args)
return await _ctx.execute(int)
[docs]
async def max_used_space(self) -> int:
"""The maximum bytes to keep in the cache without pruning.
Returns
-------
int
The `Int` scalar type represents non-fractional signed whole
numeric values. Int can represent values between -(2^31) and 2^31
- 1.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("maxUsedSpace", _args)
return await _ctx.execute(int)
[docs]
async def min_free_space(self) -> int:
"""The target amount of free disk space the garbage collector will
attempt to leave.
Returns
-------
int
The `Int` scalar type represents non-fractional signed whole
numeric values. Int can represent values between -(2^31) and 2^31
- 1.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("minFreeSpace", _args)
return await _ctx.execute(int)
[docs]
async def prune(self) -> Void | None:
"""Prune the cache of releaseable entries
Returns
-------
Void | None
The absence of a value. A Null Void is used as a placeholder for
resolvers that do not return anything.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("prune", _args)
await _ctx.execute()
[docs]
async def reserved_space(self) -> int:
"""Returns
-------
int
The `Int` scalar type represents non-fractional signed whole
numeric values. Int can represent values between -(2^31) and 2^31
- 1.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("reservedSpace", _args)
return await _ctx.execute(int)
@typecheck
class EngineCacheEntry(Type):
"""An individual cache entry in a cache entry set"""
[docs]
async def actively_used(self) -> bool:
"""Whether the cache entry is actively being used.
Returns
-------
bool
The `Boolean` scalar type represents `true` or `false`.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("activelyUsed", _args)
return await _ctx.execute(bool)
[docs]
async def created_time_unix_nano(self) -> int:
"""The time the cache entry was created, in Unix nanoseconds.
Returns
-------
int
The `Int` scalar type represents non-fractional signed whole
numeric values. Int can represent values between -(2^31) and 2^31
- 1.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("createdTimeUnixNano", _args)
return await _ctx.execute(int)
[docs]
async def description(self) -> str:
"""The description of the cache entry.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("description", _args)
return await _ctx.execute(str)
[docs]
async def disk_space_bytes(self) -> int:
"""The disk space used by the cache entry.
Returns
-------
int
The `Int` scalar type represents non-fractional signed whole
numeric values. Int can represent values between -(2^31) and 2^31
- 1.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("diskSpaceBytes", _args)
return await _ctx.execute(int)
[docs]
async def id(self) -> EngineCacheEntryID:
"""A unique identifier for this EngineCacheEntry.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
EngineCacheEntryID
The `EngineCacheEntryID` scalar type represents an identifier for
an object of type EngineCacheEntry.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(EngineCacheEntryID)
[docs]
async def most_recent_use_time_unix_nano(self) -> int:
"""The most recent time the cache entry was used, in Unix nanoseconds.
Returns
-------
int
The `Int` scalar type represents non-fractional signed whole
numeric values. Int can represent values between -(2^31) and 2^31
- 1.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("mostRecentUseTimeUnixNano", _args)
return await _ctx.execute(int)
@typecheck
class EngineCacheEntrySet(Type):
"""A set of cache entries returned by a query to a cache"""
[docs]
async def disk_space_bytes(self) -> int:
"""The total disk space used by the cache entries in this set.
Returns
-------
int
The `Int` scalar type represents non-fractional signed whole
numeric values. Int can represent values between -(2^31) and 2^31
- 1.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("diskSpaceBytes", _args)
return await _ctx.execute(int)
[docs]
async def entries(self) -> list[EngineCacheEntry]:
"""The list of individual cache entries in the set"""
_args: list[Arg] = []
_ctx = self._select("entries", _args)
_ctx = EngineCacheEntry(_ctx)._select("id", [])
@dataclass
class Response:
id: EngineCacheEntryID
_ids = await _ctx.execute(list[Response])
return [
EngineCacheEntry(
Client.from_context(_ctx)._select(
"loadEngineCacheEntryFromID",
[Arg("id", v.id)],
)
)
for v in _ids
]
[docs]
async def entry_count(self) -> int:
"""The number of cache entries in this set.
Returns
-------
int
The `Int` scalar type represents non-fractional signed whole
numeric values. Int can represent values between -(2^31) and 2^31
- 1.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("entryCount", _args)
return await _ctx.execute(int)
[docs]
async def id(self) -> EngineCacheEntrySetID:
"""A unique identifier for this EngineCacheEntrySet.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
EngineCacheEntrySetID
The `EngineCacheEntrySetID` scalar type represents an identifier
for an object of type EngineCacheEntrySet.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(EngineCacheEntrySetID)
@typecheck
class EnumTypeDef(Type):
"""A definition of a custom enum defined in a Module."""
[docs]
async def description(self) -> str:
"""A doc string for the enum, if any.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("description", _args)
return await _ctx.execute(str)
[docs]
async def id(self) -> EnumTypeDefID:
"""A unique identifier for this EnumTypeDef.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
EnumTypeDefID
The `EnumTypeDefID` scalar type represents an identifier for an
object of type EnumTypeDef.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(EnumTypeDefID)
[docs]
async def name(self) -> str:
"""The name of the enum.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("name", _args)
return await _ctx.execute(str)
[docs]
def source_map(self) -> "SourceMap":
"""The location of this enum declaration."""
_args: list[Arg] = []
_ctx = self._select("sourceMap", _args)
return SourceMap(_ctx)
[docs]
async def source_module_name(self) -> str:
"""If this EnumTypeDef is associated with a Module, the name of the
module. Unset otherwise.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("sourceModuleName", _args)
return await _ctx.execute(str)
[docs]
async def values(self) -> list["EnumValueTypeDef"]:
"""The values of the enum."""
_args: list[Arg] = []
_ctx = self._select("values", _args)
_ctx = EnumValueTypeDef(_ctx)._select("id", [])
@dataclass
class Response:
id: EnumValueTypeDefID
_ids = await _ctx.execute(list[Response])
return [
EnumValueTypeDef(
Client.from_context(_ctx)._select(
"loadEnumValueTypeDefFromID",
[Arg("id", v.id)],
)
)
for v in _ids
]
@typecheck
class EnumValueTypeDef(Type):
"""A definition of a value in a custom enum defined in a Module."""
[docs]
async def description(self) -> str:
"""A doc string for the enum value, if any.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("description", _args)
return await _ctx.execute(str)
[docs]
async def id(self) -> EnumValueTypeDefID:
"""A unique identifier for this EnumValueTypeDef.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
EnumValueTypeDefID
The `EnumValueTypeDefID` scalar type represents an identifier for
an object of type EnumValueTypeDef.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(EnumValueTypeDefID)
[docs]
async def name(self) -> str:
"""The name of the enum value.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("name", _args)
return await _ctx.execute(str)
[docs]
def source_map(self) -> "SourceMap":
"""The location of this enum value declaration."""
_args: list[Arg] = []
_ctx = self._select("sourceMap", _args)
return SourceMap(_ctx)
@typecheck
class EnvVariable(Type):
"""An environment variable name and value."""
[docs]
async def id(self) -> EnvVariableID:
"""A unique identifier for this EnvVariable.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
EnvVariableID
The `EnvVariableID` scalar type represents an identifier for an
object of type EnvVariable.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(EnvVariableID)
[docs]
async def name(self) -> str:
"""The environment variable name.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("name", _args)
return await _ctx.execute(str)
[docs]
async def value(self) -> str:
"""The environment variable value.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("value", _args)
return await _ctx.execute(str)
@typecheck
class Error(Type):
[docs]
async def id(self) -> ErrorID:
"""A unique identifier for this Error.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
ErrorID
The `ErrorID` scalar type represents an identifier for an object
of type Error.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(ErrorID)
[docs]
async def message(self) -> str:
"""A description of the error.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("message", _args)
return await _ctx.execute(str)
@typecheck
class FieldTypeDef(Type):
"""A definition of a field on a custom object defined in a Module. A
field on an object has a static value, as opposed to a function on an
object whose value is computed by invoking code (and can accept
arguments)."""
[docs]
async def description(self) -> str:
"""A doc string for the field, if any.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("description", _args)
return await _ctx.execute(str)
[docs]
async def id(self) -> FieldTypeDefID:
"""A unique identifier for this FieldTypeDef.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
FieldTypeDefID
The `FieldTypeDefID` scalar type represents an identifier for an
object of type FieldTypeDef.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(FieldTypeDefID)
[docs]
async def name(self) -> str:
"""The name of the field in lowerCamelCase format.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("name", _args)
return await _ctx.execute(str)
[docs]
def source_map(self) -> "SourceMap":
"""The location of this field declaration."""
_args: list[Arg] = []
_ctx = self._select("sourceMap", _args)
return SourceMap(_ctx)
[docs]
def type_def(self) -> "TypeDef":
"""The type of the field."""
_args: list[Arg] = []
_ctx = self._select("typeDef", _args)
return TypeDef(_ctx)
@typecheck
class File(Type):
"""A file."""
[docs]
async def contents(self) -> str:
"""Retrieves the contents of the file.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("contents", _args)
return await _ctx.execute(str)
[docs]
async def digest(
self,
*,
exclude_metadata: bool | None = False,
) -> str:
"""Return the file's digest. The format of the digest is not guaranteed
to be stable between releases of Dagger. It is guaranteed to be stable
between invocations of the same Dagger engine.
Parameters
----------
exclude_metadata:
If true, exclude metadata from the digest.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args = [
Arg("excludeMetadata", exclude_metadata, False),
]
_ctx = self._select("digest", _args)
return await _ctx.execute(str)
[docs]
async def export(
self,
path: str,
*,
allow_parent_dir_path: bool | None = False,
) -> str:
"""Writes the file to a file path on the host.
Parameters
----------
path:
Location of the written directory (e.g., "output.txt").
allow_parent_dir_path:
If allowParentDirPath is true, the path argument can be a
directory path, in which case the file will be created in that
directory.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args = [
Arg("path", path),
Arg("allowParentDirPath", allow_parent_dir_path, False),
]
_ctx = self._select("export", _args)
return await _ctx.execute(str)
[docs]
async def id(self) -> FileID:
"""A unique identifier for this File.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
FileID
The `FileID` scalar type represents an identifier for an object of
type File.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(FileID)
[docs]
async def name(self) -> str:
"""Retrieves the name of the file.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("name", _args)
return await _ctx.execute(str)
[docs]
async def size(self) -> int:
"""Retrieves the size of the file, in bytes.
Returns
-------
int
The `Int` scalar type represents non-fractional signed whole
numeric values. Int can represent values between -(2^31) and 2^31
- 1.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("size", _args)
return await _ctx.execute(int)
[docs]
async def sync(self) -> Self:
"""Force evaluation in the engine.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("sync", _args)
_id = await _ctx.execute(FileID)
_ctx = Client.from_context(_ctx)._select("loadFileFromID", [Arg("id", _id)])
return File(_ctx)
def __await__(self):
return self.sync().__await__()
[docs]
def with_name(self, name: str) -> Self:
"""Retrieves this file with its name set to the given name.
Parameters
----------
name:
Name to set file to.
"""
_args = [
Arg("name", name),
]
_ctx = self._select("withName", _args)
return File(_ctx)
[docs]
def with_timestamps(self, timestamp: int) -> Self:
"""Retrieves this file with its created/modified timestamps set to the
given time.
Parameters
----------
timestamp:
Timestamp to set dir/files in.
Formatted in seconds following Unix epoch (e.g., 1672531199).
"""
_args = [
Arg("timestamp", timestamp),
]
_ctx = self._select("withTimestamps", _args)
return File(_ctx)
[docs]
def with_(self, cb: Callable[["File"], "File"]) -> "File":
"""Call the provided callable with current File.
This is useful for reusability and readability by not breaking the calling chain.
"""
return cb(self)
@typecheck
class Function(Type):
"""Function represents a resolver provided by a Module. A function
always evaluates against a parent object and is given a set of named
arguments."""
[docs]
async def args(self) -> list["FunctionArg"]:
"""Arguments accepted by the function, if any."""
_args: list[Arg] = []
_ctx = self._select("args", _args)
_ctx = FunctionArg(_ctx)._select("id", [])
@dataclass
class Response:
id: FunctionArgID
_ids = await _ctx.execute(list[Response])
return [
FunctionArg(
Client.from_context(_ctx)._select(
"loadFunctionArgFromID",
[Arg("id", v.id)],
)
)
for v in _ids
]
[docs]
async def description(self) -> str:
"""A doc string for the function, if any.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("description", _args)
return await _ctx.execute(str)
[docs]
async def id(self) -> FunctionID:
"""A unique identifier for this Function.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
FunctionID
The `FunctionID` scalar type represents an identifier for an
object of type Function.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(FunctionID)
[docs]
async def name(self) -> str:
"""The name of the function.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("name", _args)
return await _ctx.execute(str)
[docs]
def return_type(self) -> "TypeDef":
"""The type returned by the function."""
_args: list[Arg] = []
_ctx = self._select("returnType", _args)
return TypeDef(_ctx)
[docs]
def source_map(self) -> "SourceMap":
"""The location of this function declaration."""
_args: list[Arg] = []
_ctx = self._select("sourceMap", _args)
return SourceMap(_ctx)
[docs]
def with_arg(
self,
name: str,
type_def: "TypeDef",
*,
description: str | None = "",
default_value: JSON | None = None,
default_path: str | None = "",
ignore: list[str] | None = None,
source_map: "SourceMap | None" = None,
) -> Self:
"""Returns the function with the provided argument
Parameters
----------
name:
The name of the argument
type_def:
The type of the argument
description:
A doc string for the argument, if any
default_value:
A default value to use for this argument if not explicitly set by
the caller, if any
default_path:
If the argument is a Directory or File type, default to load path
from context directory, relative to root directory.
ignore:
Patterns to ignore when loading the contextual argument value.
source_map:
"""
_args = [
Arg("name", name),
Arg("typeDef", type_def),
Arg("description", description, ""),
Arg("defaultValue", default_value, None),
Arg("defaultPath", default_path, ""),
Arg("ignore", () if ignore is None else ignore, ()),
Arg("sourceMap", source_map, None),
]
_ctx = self._select("withArg", _args)
return Function(_ctx)
[docs]
def with_description(self, description: str) -> Self:
"""Returns the function with the given doc string.
Parameters
----------
description:
The doc string to set.
"""
_args = [
Arg("description", description),
]
_ctx = self._select("withDescription", _args)
return Function(_ctx)
[docs]
def with_source_map(self, source_map: "SourceMap") -> Self:
"""Returns the function with the given source map.
Parameters
----------
source_map:
The source map for the function definition.
"""
_args = [
Arg("sourceMap", source_map),
]
_ctx = self._select("withSourceMap", _args)
return Function(_ctx)
[docs]
def with_(self, cb: Callable[["Function"], "Function"]) -> "Function":
"""Call the provided callable with current Function.
This is useful for reusability and readability by not breaking the calling chain.
"""
return cb(self)
@typecheck
class FunctionArg(Type):
"""An argument accepted by a function. This is a specification for an
argument at function definition time, not an argument passed at
function call time."""
[docs]
async def default_path(self) -> str:
"""Only applies to arguments of type File or Directory. If the argument
is not set, load it from the given path in the context directory
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("defaultPath", _args)
return await _ctx.execute(str)
[docs]
async def default_value(self) -> JSON:
"""A default value to use for this argument when not explicitly set by
the caller, if any.
Returns
-------
JSON
An arbitrary JSON-encoded value.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("defaultValue", _args)
return await _ctx.execute(JSON)
[docs]
async def description(self) -> str:
"""A doc string for the argument, if any.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("description", _args)
return await _ctx.execute(str)
[docs]
async def id(self) -> FunctionArgID:
"""A unique identifier for this FunctionArg.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
FunctionArgID
The `FunctionArgID` scalar type represents an identifier for an
object of type FunctionArg.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(FunctionArgID)
[docs]
async def ignore(self) -> list[str]:
"""Only applies to arguments of type Directory. The ignore patterns are
applied to the input directory, and matching entries are filtered out,
in a cache-efficient manner.
Returns
-------
list[str]
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("ignore", _args)
return await _ctx.execute(list[str])
[docs]
async def name(self) -> str:
"""The name of the argument in lowerCamelCase format.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("name", _args)
return await _ctx.execute(str)
[docs]
def source_map(self) -> "SourceMap":
"""The location of this arg declaration."""
_args: list[Arg] = []
_ctx = self._select("sourceMap", _args)
return SourceMap(_ctx)
[docs]
def type_def(self) -> "TypeDef":
"""The type of the argument."""
_args: list[Arg] = []
_ctx = self._select("typeDef", _args)
return TypeDef(_ctx)
@typecheck
class FunctionCall(Type):
"""An active function call."""
[docs]
async def id(self) -> FunctionCallID:
"""A unique identifier for this FunctionCall.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
FunctionCallID
The `FunctionCallID` scalar type represents an identifier for an
object of type FunctionCall.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(FunctionCallID)
[docs]
async def name(self) -> str:
"""The name of the function being called.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("name", _args)
return await _ctx.execute(str)
[docs]
async def parent(self) -> JSON:
"""The value of the parent object of the function being called. If the
function is top-level to the module, this is always an empty object.
Returns
-------
JSON
An arbitrary JSON-encoded value.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("parent", _args)
return await _ctx.execute(JSON)
[docs]
async def parent_name(self) -> str:
"""The name of the parent object of the function being called. If the
function is top-level to the module, this is the name of the module.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("parentName", _args)
return await _ctx.execute(str)
[docs]
async def return_error(self, error: Error) -> Void | None:
"""Return an error from the function.
Parameters
----------
error:
The error to return.
Returns
-------
Void | None
The absence of a value. A Null Void is used as a placeholder for
resolvers that do not return anything.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args = [
Arg("error", error),
]
_ctx = self._select("returnError", _args)
await _ctx.execute()
[docs]
async def return_value(self, value: JSON) -> Void | None:
"""Set the return value of the function call to the provided value.
Parameters
----------
value:
JSON serialization of the return value.
Returns
-------
Void | None
The absence of a value. A Null Void is used as a placeholder for
resolvers that do not return anything.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args = [
Arg("value", value),
]
_ctx = self._select("returnValue", _args)
await _ctx.execute()
@typecheck
class FunctionCallArgValue(Type):
"""A value passed as a named argument to a function call."""
[docs]
async def id(self) -> FunctionCallArgValueID:
"""A unique identifier for this FunctionCallArgValue.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
FunctionCallArgValueID
The `FunctionCallArgValueID` scalar type represents an identifier
for an object of type FunctionCallArgValue.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(FunctionCallArgValueID)
[docs]
async def name(self) -> str:
"""The name of the argument.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("name", _args)
return await _ctx.execute(str)
[docs]
async def value(self) -> JSON:
"""The value of the argument represented as a JSON serialized string.
Returns
-------
JSON
An arbitrary JSON-encoded value.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("value", _args)
return await _ctx.execute(JSON)
@typecheck
class GeneratedCode(Type):
"""The result of running an SDK's codegen."""
[docs]
def code(self) -> Directory:
"""The directory containing the generated code."""
_args: list[Arg] = []
_ctx = self._select("code", _args)
return Directory(_ctx)
[docs]
async def id(self) -> GeneratedCodeID:
"""A unique identifier for this GeneratedCode.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
GeneratedCodeID
The `GeneratedCodeID` scalar type represents an identifier for an
object of type GeneratedCode.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(GeneratedCodeID)
[docs]
async def vcs_generated_paths(self) -> list[str]:
"""List of paths to mark generated in version control (i.e.
.gitattributes).
Returns
-------
list[str]
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("vcsGeneratedPaths", _args)
return await _ctx.execute(list[str])
[docs]
async def vcs_ignored_paths(self) -> list[str]:
"""List of paths to ignore in version control (i.e. .gitignore).
Returns
-------
list[str]
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("vcsIgnoredPaths", _args)
return await _ctx.execute(list[str])
[docs]
def with_vcs_generated_paths(self, paths: list[str]) -> Self:
"""Set the list of paths to mark generated in version control."""
_args = [
Arg("paths", paths),
]
_ctx = self._select("withVCSGeneratedPaths", _args)
return GeneratedCode(_ctx)
[docs]
def with_vcs_ignored_paths(self, paths: list[str]) -> Self:
"""Set the list of paths to ignore in version control."""
_args = [
Arg("paths", paths),
]
_ctx = self._select("withVCSIgnoredPaths", _args)
return GeneratedCode(_ctx)
[docs]
def with_(
self, cb: Callable[["GeneratedCode"], "GeneratedCode"]
) -> "GeneratedCode":
"""Call the provided callable with current GeneratedCode.
This is useful for reusability and readability by not breaking the calling chain.
"""
return cb(self)
@typecheck
class GitModuleSource(Type):
"""Module source originating from a git repo."""
[docs]
async def clone_ref(self) -> str:
"""The ref to clone the root of the git repo from
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("cloneRef", _args)
return await _ctx.execute(str)
[docs]
async def commit(self) -> str:
"""The resolved commit of the git repo this source points to.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("commit", _args)
return await _ctx.execute(str)
[docs]
def context_directory(self) -> Directory:
"""The directory containing everything needed to load load and use the
module.
"""
_args: list[Arg] = []
_ctx = self._select("contextDirectory", _args)
return Directory(_ctx)
[docs]
async def html_repo_url(self) -> str:
"""The URL to access the web view of the repository (e.g., GitHub,
GitLab, Bitbucket)
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("htmlRepoURL", _args)
return await _ctx.execute(str)
[docs]
async def html_url(self) -> str:
"""The URL to the source's git repo in a web browser
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("htmlURL", _args)
return await _ctx.execute(str)
[docs]
async def id(self) -> GitModuleSourceID:
"""A unique identifier for this GitModuleSource.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
GitModuleSourceID
The `GitModuleSourceID` scalar type represents an identifier for
an object of type GitModuleSource.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(GitModuleSourceID)
[docs]
async def root(self) -> str:
"""The clean module name of the root of the module
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("root", _args)
return await _ctx.execute(str)
[docs]
async def root_subpath(self) -> str:
"""The path to the root of the module source under the context directory.
This directory contains its configuration file. It also contains its
source code (possibly as a subdirectory).
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("rootSubpath", _args)
return await _ctx.execute(str)
[docs]
async def version(self) -> str:
"""The specified version of the git repo this source points to.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("version", _args)
return await _ctx.execute(str)
@typecheck
class GitRef(Type):
"""A git ref (tag, branch, or commit)."""
[docs]
async def commit(self) -> str:
"""The resolved commit id at this ref.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("commit", _args)
return await _ctx.execute(str)
[docs]
async def id(self) -> GitRefID:
"""A unique identifier for this GitRef.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
GitRefID
The `GitRefID` scalar type represents an identifier for an object
of type GitRef.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(GitRefID)
[docs]
def tree(
self,
*,
discard_git_dir: bool | None = False,
) -> Directory:
"""The filesystem tree at this ref.
Parameters
----------
discard_git_dir:
Set to true to discard .git directory.
"""
_args = [
Arg("discardGitDir", discard_git_dir, False),
]
_ctx = self._select("tree", _args)
return Directory(_ctx)
@typecheck
class GitRepository(Type):
"""A git repository."""
[docs]
def branch(self, name: str) -> GitRef:
"""Returns details of a branch.
Parameters
----------
name:
Branch's name (e.g., "main").
"""
_args = [
Arg("name", name),
]
_ctx = self._select("branch", _args)
return GitRef(_ctx)
[docs]
def commit(self, id: str) -> GitRef:
"""Returns details of a commit.
Parameters
----------
id:
Identifier of the commit (e.g.,
"b6315d8f2810962c601af73f86831f6866ea798b").
"""
_args = [
Arg("id", id),
]
_ctx = self._select("commit", _args)
return GitRef(_ctx)
[docs]
def head(self) -> GitRef:
"""Returns details for HEAD."""
_args: list[Arg] = []
_ctx = self._select("head", _args)
return GitRef(_ctx)
[docs]
async def id(self) -> GitRepositoryID:
"""A unique identifier for this GitRepository.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
GitRepositoryID
The `GitRepositoryID` scalar type represents an identifier for an
object of type GitRepository.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(GitRepositoryID)
[docs]
def ref(self, name: str) -> GitRef:
"""Returns details of a ref.
Parameters
----------
name:
Ref's name (can be a commit identifier, a tag name, a branch name,
or a fully-qualified ref).
"""
_args = [
Arg("name", name),
]
_ctx = self._select("ref", _args)
return GitRef(_ctx)
[docs]
def tag(self, name: str) -> GitRef:
"""Returns details of a tag.
Parameters
----------
name:
Tag's name (e.g., "v0.3.9").
"""
_args = [
Arg("name", name),
]
_ctx = self._select("tag", _args)
return GitRef(_ctx)
[docs]
def with_auth_token(self, token: "Secret") -> Self:
"""Token to authenticate the remote with.
Parameters
----------
token:
Secret used to populate the password during basic HTTP
Authorization
"""
_args = [
Arg("token", token),
]
_ctx = self._select("withAuthToken", _args)
return GitRepository(_ctx)
[docs]
def with_(
self, cb: Callable[["GitRepository"], "GitRepository"]
) -> "GitRepository":
"""Call the provided callable with current GitRepository.
This is useful for reusability and readability by not breaking the calling chain.
"""
return cb(self)
@typecheck
class Host(Type):
"""Information about the host environment."""
[docs]
def directory(
self,
path: str,
*,
exclude: list[str] | None = None,
include: list[str] | None = None,
) -> Directory:
"""Accesses a directory on the host.
Parameters
----------
path:
Location of the directory to access (e.g., ".").
exclude:
Exclude artifacts that match the given pattern (e.g.,
["node_modules/", ".git*"]).
include:
Include only artifacts that match the given pattern (e.g.,
["app/", "package.*"]).
"""
_args = [
Arg("path", path),
Arg("exclude", () if exclude is None else exclude, ()),
Arg("include", () if include is None else include, ()),
]
_ctx = self._select("directory", _args)
return Directory(_ctx)
[docs]
def file(self, path: str) -> File:
"""Accesses a file on the host.
Parameters
----------
path:
Location of the file to retrieve (e.g., "README.md").
"""
_args = [
Arg("path", path),
]
_ctx = self._select("file", _args)
return File(_ctx)
[docs]
async def id(self) -> HostID:
"""A unique identifier for this Host.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
HostID
The `HostID` scalar type represents an identifier for an object of
type Host.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(HostID)
[docs]
def service(
self,
ports: list[PortForward],
*,
host: str | None = "localhost",
) -> "Service":
"""Creates a service that forwards traffic to a specified address via the
host.
Parameters
----------
ports:
Ports to expose via the service, forwarding through the host
network.
If a port's frontend is unspecified or 0, it defaults to the same
as the backend port.
An empty set of ports is not valid; an error will be returned.
host:
Upstream host to forward traffic to.
"""
_args = [
Arg("ports", ports),
Arg("host", host, "localhost"),
]
_ctx = self._select("service", _args)
return Service(_ctx)
[docs]
def set_secret_file(self, name: str, path: str) -> "Secret":
"""Sets a secret given a user-defined name and the file path on the host,
and returns the secret.
The file is limited to a size of 512000 bytes.
Parameters
----------
name:
The user defined name for this secret.
path:
Location of the file to set as a secret.
"""
_args = [
Arg("name", name),
Arg("path", path),
]
_ctx = self._select("setSecretFile", _args)
return Secret(_ctx)
[docs]
def tunnel(
self,
service: "Service",
*,
ports: list[PortForward] | None = None,
native: bool | None = False,
) -> "Service":
"""Creates a tunnel that forwards traffic from the host to a service.
Parameters
----------
service:
Service to send traffic from the tunnel.
ports:
Configure explicit port forwarding rules for the tunnel.
If a port's frontend is unspecified or 0, a random port will be
chosen by the host.
If no ports are given, all of the service's ports are forwarded.
If native is true, each port maps to the same port on the host. If
native is false, each port maps to a random port chosen by the
host.
If ports are given and native is true, the ports are additive.
native:
Map each service port to the same port on the host, as if the
service were running natively.
Note: enabling may result in port conflicts.
"""
_args = [
Arg("service", service),
Arg("ports", () if ports is None else ports, ()),
Arg("native", native, False),
]
_ctx = self._select("tunnel", _args)
return Service(_ctx)
[docs]
def unix_socket(self, path: str) -> "Socket":
"""Accesses a Unix socket on the host.
Parameters
----------
path:
Location of the Unix socket (e.g., "/var/run/docker.sock").
"""
_args = [
Arg("path", path),
]
_ctx = self._select("unixSocket", _args)
return Socket(_ctx)
@typecheck
class InputTypeDef(Type):
"""A graphql input type, which is essentially just a group of named
args. This is currently only used to represent pre-existing usage of
graphql input types in the core API. It is not used by user modules
and shouldn't ever be as user module accept input objects via their id
rather than graphql input types."""
@typecheck
class InterfaceTypeDef(Type):
"""A definition of a custom interface defined in a Module."""
[docs]
async def description(self) -> str:
"""The doc string for the interface, if any.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("description", _args)
return await _ctx.execute(str)
[docs]
async def functions(self) -> list[Function]:
"""Functions defined on this interface, if any."""
_args: list[Arg] = []
_ctx = self._select("functions", _args)
_ctx = Function(_ctx)._select("id", [])
@dataclass
class Response:
id: FunctionID
_ids = await _ctx.execute(list[Response])
return [
Function(
Client.from_context(_ctx)._select(
"loadFunctionFromID",
[Arg("id", v.id)],
)
)
for v in _ids
]
[docs]
async def id(self) -> InterfaceTypeDefID:
"""A unique identifier for this InterfaceTypeDef.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
InterfaceTypeDefID
The `InterfaceTypeDefID` scalar type represents an identifier for
an object of type InterfaceTypeDef.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(InterfaceTypeDefID)
[docs]
async def name(self) -> str:
"""The name of the interface.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("name", _args)
return await _ctx.execute(str)
[docs]
def source_map(self) -> "SourceMap":
"""The location of this interface declaration."""
_args: list[Arg] = []
_ctx = self._select("sourceMap", _args)
return SourceMap(_ctx)
[docs]
async def source_module_name(self) -> str:
"""If this InterfaceTypeDef is associated with a Module, the name of the
module. Unset otherwise.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("sourceModuleName", _args)
return await _ctx.execute(str)
@typecheck
class Label(Type):
"""A simple key value object that represents a label."""
[docs]
async def id(self) -> LabelID:
"""A unique identifier for this Label.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
LabelID
The `LabelID` scalar type represents an identifier for an object
of type Label.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(LabelID)
[docs]
async def name(self) -> str:
"""The label name.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("name", _args)
return await _ctx.execute(str)
[docs]
async def value(self) -> str:
"""The label value.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("value", _args)
return await _ctx.execute(str)
@typecheck
class ListTypeDef(Type):
"""A definition of a list type in a Module."""
[docs]
def element_type_def(self) -> "TypeDef":
"""The type of the elements in the list."""
_args: list[Arg] = []
_ctx = self._select("elementTypeDef", _args)
return TypeDef(_ctx)
[docs]
async def id(self) -> ListTypeDefID:
"""A unique identifier for this ListTypeDef.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
ListTypeDefID
The `ListTypeDefID` scalar type represents an identifier for an
object of type ListTypeDef.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(ListTypeDefID)
@typecheck
class LocalModuleSource(Type):
"""Module source that that originates from a path locally relative to
an arbitrary directory."""
[docs]
def context_directory(self) -> Directory:
"""The directory containing everything needed to load load and use the
module.
"""
_args: list[Arg] = []
_ctx = self._select("contextDirectory", _args)
return Directory(_ctx)
[docs]
async def id(self) -> LocalModuleSourceID:
"""A unique identifier for this LocalModuleSource.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
LocalModuleSourceID
The `LocalModuleSourceID` scalar type represents an identifier for
an object of type LocalModuleSource.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(LocalModuleSourceID)
[docs]
async def rel_host_path(self) -> str:
"""The relative path to the module root from the host directory
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("relHostPath", _args)
return await _ctx.execute(str)
[docs]
async def root_subpath(self) -> str:
"""The path to the root of the module source under the context directory.
This directory contains its configuration file. It also contains its
source code (possibly as a subdirectory).
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("rootSubpath", _args)
return await _ctx.execute(str)
@typecheck
class Module(Type):
"""A Dagger module."""
[docs]
async def dependencies(self) -> list["Module"]:
"""Modules used by this module."""
_args: list[Arg] = []
_ctx = self._select("dependencies", _args)
_ctx = Module(_ctx)._select("id", [])
@dataclass
class Response:
id: ModuleID
_ids = await _ctx.execute(list[Response])
return [
Module(
Client.from_context(_ctx)._select(
"loadModuleFromID",
[Arg("id", v.id)],
)
)
for v in _ids
]
[docs]
async def dependency_config(self) -> list["ModuleDependency"]:
"""The dependencies as configured by the module."""
_args: list[Arg] = []
_ctx = self._select("dependencyConfig", _args)
_ctx = ModuleDependency(_ctx)._select("id", [])
@dataclass
class Response:
id: ModuleDependencyID
_ids = await _ctx.execute(list[Response])
return [
ModuleDependency(
Client.from_context(_ctx)._select(
"loadModuleDependencyFromID",
[Arg("id", v.id)],
)
)
for v in _ids
]
[docs]
async def description(self) -> str:
"""The doc string of the module, if any
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("description", _args)
return await _ctx.execute(str)
[docs]
async def enums(self) -> list["TypeDef"]:
"""Enumerations served by this module."""
_args: list[Arg] = []
_ctx = self._select("enums", _args)
_ctx = TypeDef(_ctx)._select("id", [])
@dataclass
class Response:
id: TypeDefID
_ids = await _ctx.execute(list[Response])
return [
TypeDef(
Client.from_context(_ctx)._select(
"loadTypeDefFromID",
[Arg("id", v.id)],
)
)
for v in _ids
]
[docs]
def generated_context_diff(self) -> Directory:
"""The generated files and directories made on top of the module source's
context directory.
"""
_args: list[Arg] = []
_ctx = self._select("generatedContextDiff", _args)
return Directory(_ctx)
[docs]
def generated_context_directory(self) -> Directory:
"""The module source's context plus any configuration and source files
created by codegen.
"""
_args: list[Arg] = []
_ctx = self._select("generatedContextDirectory", _args)
return Directory(_ctx)
[docs]
async def id(self) -> ModuleID:
"""A unique identifier for this Module.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
ModuleID
The `ModuleID` scalar type represents an identifier for an object
of type Module.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(ModuleID)
[docs]
def initialize(self) -> Self:
"""Retrieves the module with the objects loaded via its SDK."""
_args: list[Arg] = []
_ctx = self._select("initialize", _args)
return Module(_ctx)
[docs]
async def interfaces(self) -> list["TypeDef"]:
"""Interfaces served by this module."""
_args: list[Arg] = []
_ctx = self._select("interfaces", _args)
_ctx = TypeDef(_ctx)._select("id", [])
@dataclass
class Response:
id: TypeDefID
_ids = await _ctx.execute(list[Response])
return [
TypeDef(
Client.from_context(_ctx)._select(
"loadTypeDefFromID",
[Arg("id", v.id)],
)
)
for v in _ids
]
[docs]
async def name(self) -> str:
"""The name of the module
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("name", _args)
return await _ctx.execute(str)
[docs]
async def objects(self) -> list["TypeDef"]:
"""Objects served by this module."""
_args: list[Arg] = []
_ctx = self._select("objects", _args)
_ctx = TypeDef(_ctx)._select("id", [])
@dataclass
class Response:
id: TypeDefID
_ids = await _ctx.execute(list[Response])
return [
TypeDef(
Client.from_context(_ctx)._select(
"loadTypeDefFromID",
[Arg("id", v.id)],
)
)
for v in _ids
]
[docs]
def runtime(self) -> Container:
"""The container that runs the module's entrypoint. It will fail to
execute if the module doesn't compile.
"""
_args: list[Arg] = []
_ctx = self._select("runtime", _args)
return Container(_ctx)
[docs]
async def sdk(self) -> str:
"""The SDK used by this module. Either a name of a builtin SDK or a
module source ref string pointing to the SDK's implementation.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("sdk", _args)
return await _ctx.execute(str)
[docs]
async def serve(self) -> Void | None:
"""Serve a module's API in the current session.
Note: this can only be called once per session. In the future, it
could return a stream or service to remove the side effect.
Returns
-------
Void | None
The absence of a value. A Null Void is used as a placeholder for
resolvers that do not return anything.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("serve", _args)
await _ctx.execute()
[docs]
def source(self) -> "ModuleSource":
"""The source for the module."""
_args: list[Arg] = []
_ctx = self._select("source", _args)
return ModuleSource(_ctx)
[docs]
def with_description(self, description: str) -> Self:
"""Retrieves the module with the given description
Parameters
----------
description:
The description to set
"""
_args = [
Arg("description", description),
]
_ctx = self._select("withDescription", _args)
return Module(_ctx)
[docs]
def with_enum(self, enum: "TypeDef") -> Self:
"""This module plus the given Enum type and associated values"""
_args = [
Arg("enum", enum),
]
_ctx = self._select("withEnum", _args)
return Module(_ctx)
[docs]
def with_interface(self, iface: "TypeDef") -> Self:
"""This module plus the given Interface type and associated functions"""
_args = [
Arg("iface", iface),
]
_ctx = self._select("withInterface", _args)
return Module(_ctx)
[docs]
def with_object(self, object: "TypeDef") -> Self:
"""This module plus the given Object type and associated functions."""
_args = [
Arg("object", object),
]
_ctx = self._select("withObject", _args)
return Module(_ctx)
[docs]
def with_source(
self,
source: "ModuleSource",
*,
engine_version: str | None = None,
) -> Self:
"""Retrieves the module with basic configuration loaded if present.
Parameters
----------
source:
The module source to initialize from.
engine_version:
The engine version to upgrade to.
"""
_args = [
Arg("source", source),
Arg("engineVersion", engine_version, None),
]
_ctx = self._select("withSource", _args)
return Module(_ctx)
[docs]
def with_(self, cb: Callable[["Module"], "Module"]) -> "Module":
"""Call the provided callable with current Module.
This is useful for reusability and readability by not breaking the calling chain.
"""
return cb(self)
@typecheck
class ModuleDependency(Type):
"""The configuration of dependency of a module."""
[docs]
async def id(self) -> ModuleDependencyID:
"""A unique identifier for this ModuleDependency.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
ModuleDependencyID
The `ModuleDependencyID` scalar type represents an identifier for
an object of type ModuleDependency.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(ModuleDependencyID)
[docs]
async def name(self) -> str:
"""The name of the dependency module.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("name", _args)
return await _ctx.execute(str)
[docs]
def source(self) -> "ModuleSource":
"""The source for the dependency module."""
_args: list[Arg] = []
_ctx = self._select("source", _args)
return ModuleSource(_ctx)
@typecheck
class ModuleSource(Type):
"""The source needed to load and run a module, along with any metadata
about the source such as versions/urls/etc."""
[docs]
def as_git_source(self) -> GitModuleSource:
"""If the source is a of kind git, the git source representation of it."""
_args: list[Arg] = []
_ctx = self._select("asGitSource", _args)
return GitModuleSource(_ctx)
[docs]
def as_local_source(self) -> LocalModuleSource:
"""If the source is of kind local, the local source representation of it."""
_args: list[Arg] = []
_ctx = self._select("asLocalSource", _args)
return LocalModuleSource(_ctx)
[docs]
def as_module(
self,
*,
engine_version: str | None = None,
) -> Module:
"""Load the source as a module. If this is a local source, the parent
directory must have been provided during module source creation
Parameters
----------
engine_version:
The engine version to upgrade to.
"""
_args = [
Arg("engineVersion", engine_version, None),
]
_ctx = self._select("asModule", _args)
return Module(_ctx)
[docs]
async def as_string(self) -> str:
"""A human readable ref string representation of this module source.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("asString", _args)
return await _ctx.execute(str)
[docs]
async def config_exists(self) -> bool:
"""Returns whether the module source has a configuration file.
Returns
-------
bool
The `Boolean` scalar type represents `true` or `false`.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("configExists", _args)
return await _ctx.execute(bool)
[docs]
def context_directory(self) -> Directory:
"""The directory containing everything needed to load and use the module."""
_args: list[Arg] = []
_ctx = self._select("contextDirectory", _args)
return Directory(_ctx)
[docs]
async def dependencies(self) -> list[ModuleDependency]:
"""The effective module source dependencies from the configuration, and
calls to withDependencies and withoutDependencies.
"""
_args: list[Arg] = []
_ctx = self._select("dependencies", _args)
_ctx = ModuleDependency(_ctx)._select("id", [])
@dataclass
class Response:
id: ModuleDependencyID
_ids = await _ctx.execute(list[Response])
return [
ModuleDependency(
Client.from_context(_ctx)._select(
"loadModuleDependencyFromID",
[Arg("id", v.id)],
)
)
for v in _ids
]
[docs]
async def digest(self) -> str:
"""Return the module source's content digest. The format of the digest is
not guaranteed to be stable between releases of Dagger. It is
guaranteed to be stable between invocations of the same Dagger engine.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("digest", _args)
return await _ctx.execute(str)
[docs]
def directory(self, path: str) -> Directory:
"""The directory containing the module configuration and source code
(source code may be in a subdir).
Parameters
----------
path:
The path from the source directory to select.
"""
_args = [
Arg("path", path),
]
_ctx = self._select("directory", _args)
return Directory(_ctx)
[docs]
async def id(self) -> ModuleSourceID:
"""A unique identifier for this ModuleSource.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
ModuleSourceID
The `ModuleSourceID` scalar type represents an identifier for an
object of type ModuleSource.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(ModuleSourceID)
[docs]
async def kind(self) -> ModuleSourceKind:
"""The kind of source (e.g. local, git, etc.)
Returns
-------
ModuleSourceKind
The kind of module source.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("kind", _args)
return await _ctx.execute(ModuleSourceKind)
[docs]
async def module_name(self) -> str:
"""If set, the name of the module this source references, including any
overrides at runtime by callers.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("moduleName", _args)
return await _ctx.execute(str)
[docs]
async def module_original_name(self) -> str:
"""The original name of the module this source references, as defined in
the module configuration.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("moduleOriginalName", _args)
return await _ctx.execute(str)
[docs]
async def pin(self) -> str:
"""The pinned version of this module source.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("pin", _args)
return await _ctx.execute(str)
[docs]
async def resolve_context_path_from_caller(self) -> str:
"""The path to the module source's context directory on the caller's
filesystem. Only valid for local sources.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("resolveContextPathFromCaller", _args)
return await _ctx.execute(str)
[docs]
def resolve_dependency(self, dep: Self) -> Self:
"""Resolve the provided module source arg as a dependency relative to
this module source.
Parameters
----------
dep:
The dependency module source to resolve.
"""
_args = [
Arg("dep", dep),
]
_ctx = self._select("resolveDependency", _args)
return ModuleSource(_ctx)
[docs]
def resolve_directory_from_caller(
self,
path: str,
*,
view_name: str | None = None,
ignore: list[str] | None = None,
) -> Directory:
"""Load a directory from the caller optionally with a given view applied.
Parameters
----------
path:
The path on the caller's filesystem to load.
view_name:
If set, the name of the view to apply to the path.
ignore:
Patterns to ignore when loading the directory.
"""
_args = [
Arg("path", path),
Arg("viewName", view_name, None),
Arg("ignore", () if ignore is None else ignore, ()),
]
_ctx = self._select("resolveDirectoryFromCaller", _args)
return Directory(_ctx)
[docs]
def resolve_from_caller(self) -> Self:
"""Load the source from its path on the caller's filesystem, including
only needed+configured files and directories. Only valid for local
sources.
"""
_args: list[Arg] = []
_ctx = self._select("resolveFromCaller", _args)
return ModuleSource(_ctx)
[docs]
async def source_root_subpath(self) -> str:
"""The path relative to context of the root of the module source, which
contains dagger.json. It also contains the module implementation
source code, but that may or may not being a subdir of this root.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("sourceRootSubpath", _args)
return await _ctx.execute(str)
[docs]
async def source_subpath(self) -> str:
"""The path relative to context of the module implementation source code.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("sourceSubpath", _args)
return await _ctx.execute(str)
[docs]
def view(self, name: str) -> "ModuleSourceView":
"""Retrieve a named view defined for this module source.
Parameters
----------
name:
The name of the view to retrieve.
"""
_args = [
Arg("name", name),
]
_ctx = self._select("view", _args)
return ModuleSourceView(_ctx)
[docs]
async def views(self) -> list["ModuleSourceView"]:
"""The named views defined for this module source, which are sets of
directory filters that can be applied to directory arguments provided
to functions.
"""
_args: list[Arg] = []
_ctx = self._select("views", _args)
_ctx = ModuleSourceView(_ctx)._select("id", [])
@dataclass
class Response:
id: ModuleSourceViewID
_ids = await _ctx.execute(list[Response])
return [
ModuleSourceView(
Client.from_context(_ctx)._select(
"loadModuleSourceViewFromID",
[Arg("id", v.id)],
)
)
for v in _ids
]
[docs]
def with_context_directory(self, dir: Directory) -> Self:
"""Update the module source with a new context directory. Only valid for
local sources.
Parameters
----------
dir:
The directory to set as the context directory.
"""
_args = [
Arg("dir", dir),
]
_ctx = self._select("withContextDirectory", _args)
return ModuleSource(_ctx)
[docs]
def with_dependencies(
self,
dependencies: list[ModuleDependency],
) -> Self:
"""Append the provided dependencies to the module source's dependency
list.
Parameters
----------
dependencies:
The dependencies to append.
"""
_args = [
Arg("dependencies", dependencies),
]
_ctx = self._select("withDependencies", _args)
return ModuleSource(_ctx)
[docs]
def with_init(self, *, merge: bool | None = False) -> Self:
"""Sets module init arguments
Parameters
----------
merge:
Merge module dependencies into the current project's
"""
_args = [
Arg("merge", merge, False),
]
_ctx = self._select("withInit", _args)
return ModuleSource(_ctx)
[docs]
def with_name(self, name: str) -> Self:
"""Update the module source with a new name.
Parameters
----------
name:
The name to set.
"""
_args = [
Arg("name", name),
]
_ctx = self._select("withName", _args)
return ModuleSource(_ctx)
[docs]
def with_sdk(self, sdk: str) -> Self:
"""Update the module source with a new SDK.
Parameters
----------
sdk:
The SDK to set.
"""
_args = [
Arg("sdk", sdk),
]
_ctx = self._select("withSDK", _args)
return ModuleSource(_ctx)
[docs]
def with_source_subpath(self, path: str) -> Self:
"""Update the module source with a new source subpath.
Parameters
----------
path:
The path to set as the source subpath.
"""
_args = [
Arg("path", path),
]
_ctx = self._select("withSourceSubpath", _args)
return ModuleSource(_ctx)
[docs]
def with_update_dependencies(self, dependencies: list[str]) -> Self:
"""Update one or more module dependencies.
Parameters
----------
dependencies:
The dependencies to update.
"""
_args = [
Arg("dependencies", dependencies),
]
_ctx = self._select("withUpdateDependencies", _args)
return ModuleSource(_ctx)
[docs]
def with_view(self, name: str, patterns: list[str]) -> Self:
"""Update the module source with a new named view.
Parameters
----------
name:
The name of the view to set.
patterns:
The patterns to set as the view filters.
"""
_args = [
Arg("name", name),
Arg("patterns", patterns),
]
_ctx = self._select("withView", _args)
return ModuleSource(_ctx)
[docs]
def without_dependencies(self, dependencies: list[str]) -> Self:
"""Remove the provided dependencies from the module source's dependency
list.
Parameters
----------
dependencies:
The dependencies to remove.
"""
_args = [
Arg("dependencies", dependencies),
]
_ctx = self._select("withoutDependencies", _args)
return ModuleSource(_ctx)
[docs]
def with_(self, cb: Callable[["ModuleSource"], "ModuleSource"]) -> "ModuleSource":
"""Call the provided callable with current ModuleSource.
This is useful for reusability and readability by not breaking the calling chain.
"""
return cb(self)
@typecheck
class ModuleSourceView(Type):
"""A named set of path filters that can be applied to directory
arguments provided to functions."""
[docs]
async def id(self) -> ModuleSourceViewID:
"""A unique identifier for this ModuleSourceView.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
ModuleSourceViewID
The `ModuleSourceViewID` scalar type represents an identifier for
an object of type ModuleSourceView.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(ModuleSourceViewID)
[docs]
async def name(self) -> str:
"""The name of the view
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("name", _args)
return await _ctx.execute(str)
[docs]
async def patterns(self) -> list[str]:
"""The patterns of the view used to filter paths
Returns
-------
list[str]
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("patterns", _args)
return await _ctx.execute(list[str])
@typecheck
class ObjectTypeDef(Type):
"""A definition of a custom object defined in a Module."""
[docs]
def constructor(self) -> Function:
"""The function used to construct new instances of this object, if any"""
_args: list[Arg] = []
_ctx = self._select("constructor", _args)
return Function(_ctx)
[docs]
async def description(self) -> str:
"""The doc string for the object, if any.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("description", _args)
return await _ctx.execute(str)
[docs]
async def fields(self) -> list[FieldTypeDef]:
"""Static fields defined on this object, if any."""
_args: list[Arg] = []
_ctx = self._select("fields", _args)
_ctx = FieldTypeDef(_ctx)._select("id", [])
@dataclass
class Response:
id: FieldTypeDefID
_ids = await _ctx.execute(list[Response])
return [
FieldTypeDef(
Client.from_context(_ctx)._select(
"loadFieldTypeDefFromID",
[Arg("id", v.id)],
)
)
for v in _ids
]
[docs]
async def functions(self) -> list[Function]:
"""Functions defined on this object, if any."""
_args: list[Arg] = []
_ctx = self._select("functions", _args)
_ctx = Function(_ctx)._select("id", [])
@dataclass
class Response:
id: FunctionID
_ids = await _ctx.execute(list[Response])
return [
Function(
Client.from_context(_ctx)._select(
"loadFunctionFromID",
[Arg("id", v.id)],
)
)
for v in _ids
]
[docs]
async def id(self) -> ObjectTypeDefID:
"""A unique identifier for this ObjectTypeDef.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
ObjectTypeDefID
The `ObjectTypeDefID` scalar type represents an identifier for an
object of type ObjectTypeDef.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(ObjectTypeDefID)
[docs]
async def name(self) -> str:
"""The name of the object.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("name", _args)
return await _ctx.execute(str)
[docs]
def source_map(self) -> "SourceMap":
"""The location of this object declaration."""
_args: list[Arg] = []
_ctx = self._select("sourceMap", _args)
return SourceMap(_ctx)
[docs]
async def source_module_name(self) -> str:
"""If this ObjectTypeDef is associated with a Module, the name of the
module. Unset otherwise.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("sourceModuleName", _args)
return await _ctx.execute(str)
@typecheck
class Port(Type):
"""A port exposed by a container."""
[docs]
async def description(self) -> str | None:
"""The port description.
Returns
-------
str | None
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("description", _args)
return await _ctx.execute(str | None)
[docs]
async def experimental_skip_healthcheck(self) -> bool:
"""Skip the health check when run as a service.
Returns
-------
bool
The `Boolean` scalar type represents `true` or `false`.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("experimentalSkipHealthcheck", _args)
return await _ctx.execute(bool)
[docs]
async def id(self) -> PortID:
"""A unique identifier for this Port.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
PortID
The `PortID` scalar type represents an identifier for an object of
type Port.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(PortID)
[docs]
async def port(self) -> int:
"""The port number.
Returns
-------
int
The `Int` scalar type represents non-fractional signed whole
numeric values. Int can represent values between -(2^31) and 2^31
- 1.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("port", _args)
return await _ctx.execute(int)
[docs]
async def protocol(self) -> NetworkProtocol:
"""The transport layer protocol.
Returns
-------
NetworkProtocol
Transport layer network protocol associated to a port.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("protocol", _args)
return await _ctx.execute(NetworkProtocol)
@typecheck
class Client(Root):
"""The root of the DAG."""
[docs]
def blob(self, digest: str) -> Directory:
"""Retrieves a content-addressed blob.
Parameters
----------
digest:
Digest of the blob
"""
_args = [
Arg("digest", digest),
]
_ctx = self._select("blob", _args)
return Directory(_ctx)
[docs]
def builtin_container(self, digest: str) -> Container:
"""Retrieves a container builtin to the engine.
Parameters
----------
digest:
Digest of the image manifest
"""
_args = [
Arg("digest", digest),
]
_ctx = self._select("builtinContainer", _args)
return Container(_ctx)
[docs]
def cache_volume(
self,
key: str,
*,
namespace: str | None = "",
) -> CacheVolume:
"""Constructs a cache volume for a given cache key.
Parameters
----------
key:
A string identifier to target this cache volume (e.g., "modules-
cache").
namespace:
"""
_args = [
Arg("key", key),
Arg("namespace", namespace, ""),
]
_ctx = self._select("cacheVolume", _args)
return CacheVolume(_ctx)
[docs]
def container(
self,
*,
platform: Platform | None = None,
) -> Container:
"""Creates a scratch container.
Optional platform argument initializes new containers to execute and
publish as that platform. Platform defaults to that of the builder's
host.
Parameters
----------
platform:
Platform to initialize the container with.
"""
_args = [
Arg("platform", platform, None),
]
_ctx = self._select("container", _args)
return Container(_ctx)
[docs]
def current_function_call(self) -> FunctionCall:
"""The FunctionCall context that the SDK caller is currently executing
in.
If the caller is not currently executing in a function, this will
return an error.
"""
_args: list[Arg] = []
_ctx = self._select("currentFunctionCall", _args)
return FunctionCall(_ctx)
[docs]
def current_module(self) -> CurrentModule:
"""The module currently being served in the session, if any."""
_args: list[Arg] = []
_ctx = self._select("currentModule", _args)
return CurrentModule(_ctx)
[docs]
async def current_type_defs(self) -> list["TypeDef"]:
"""The TypeDef representations of the objects currently being served in
the session.
"""
_args: list[Arg] = []
_ctx = self._select("currentTypeDefs", _args)
_ctx = TypeDef(_ctx)._select("id", [])
@dataclass
class Response:
id: TypeDefID
_ids = await _ctx.execute(list[Response])
return [
TypeDef(
Client.from_context(_ctx)._select(
"loadTypeDefFromID",
[Arg("id", v.id)],
)
)
for v in _ids
]
[docs]
def directory(self) -> Directory:
"""Creates an empty directory."""
_args: list[Arg] = []
_ctx = self._select("directory", _args)
return Directory(_ctx)
[docs]
def engine(self) -> Engine:
"""The Dagger engine container configuration and state"""
_args: list[Arg] = []
_ctx = self._select("engine", _args)
return Engine(_ctx)
[docs]
def error(self, message: str) -> Error:
"""Create a new error.
Parameters
----------
message:
A brief description of the error.
"""
_args = [
Arg("message", message),
]
_ctx = self._select("error", _args)
return Error(_ctx)
[docs]
def function(self, name: str, return_type: "TypeDef") -> Function:
"""Creates a function.
Parameters
----------
name:
Name of the function, in its original format from the
implementation language.
return_type:
Return type of the function.
"""
_args = [
Arg("name", name),
Arg("returnType", return_type),
]
_ctx = self._select("function", _args)
return Function(_ctx)
[docs]
def generated_code(self, code: Directory) -> GeneratedCode:
"""Create a code generation result, given a directory containing the
generated code.
"""
_args = [
Arg("code", code),
]
_ctx = self._select("generatedCode", _args)
return GeneratedCode(_ctx)
[docs]
def git(
self,
url: str,
*,
keep_git_dir: bool | None = True,
experimental_service_host: "Service | None" = None,
ssh_known_hosts: str | None = "",
ssh_auth_socket: "Socket | None" = None,
) -> GitRepository:
"""Queries a Git repository.
Parameters
----------
url:
URL of the git repository.
Can be formatted as `https://{host}/{owner}/{repo}`,
`git@{host}:{owner}/{repo}`.
Suffix ".git" is optional.
keep_git_dir:
DEPRECATED: Set to true to keep .git directory.
experimental_service_host:
A service which must be started before the repo is fetched.
ssh_known_hosts:
Set SSH known hosts
ssh_auth_socket:
Set SSH auth socket
"""
_args = [
Arg("url", url),
Arg("keepGitDir", keep_git_dir, True),
Arg("experimentalServiceHost", experimental_service_host, None),
Arg("sshKnownHosts", ssh_known_hosts, ""),
Arg("sshAuthSocket", ssh_auth_socket, None),
]
_ctx = self._select("git", _args)
return GitRepository(_ctx)
[docs]
def host(self) -> Host:
"""Queries the host environment."""
_args: list[Arg] = []
_ctx = self._select("host", _args)
return Host(_ctx)
[docs]
def http(
self,
url: str,
*,
experimental_service_host: "Service | None" = None,
) -> File:
"""Returns a file containing an http remote url content.
Parameters
----------
url:
HTTP url to get the content from (e.g., "https://docs.dagger.io").
experimental_service_host:
A service which must be started before the URL is fetched.
"""
_args = [
Arg("url", url),
Arg("experimentalServiceHost", experimental_service_host, None),
]
_ctx = self._select("http", _args)
return File(_ctx)
[docs]
def load_cache_volume_from_id(self, id: CacheVolumeID) -> CacheVolume:
"""Load a CacheVolume from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadCacheVolumeFromID", _args)
return CacheVolume(_ctx)
[docs]
def load_container_from_id(self, id: ContainerID) -> Container:
"""Load a Container from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadContainerFromID", _args)
return Container(_ctx)
[docs]
def load_current_module_from_id(self, id: CurrentModuleID) -> CurrentModule:
"""Load a CurrentModule from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadCurrentModuleFromID", _args)
return CurrentModule(_ctx)
[docs]
def load_directory_from_id(self, id: DirectoryID) -> Directory:
"""Load a Directory from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadDirectoryFromID", _args)
return Directory(_ctx)
[docs]
def load_engine_cache_entry_from_id(
self, id: EngineCacheEntryID
) -> EngineCacheEntry:
"""Load a EngineCacheEntry from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadEngineCacheEntryFromID", _args)
return EngineCacheEntry(_ctx)
[docs]
def load_engine_cache_entry_set_from_id(
self, id: EngineCacheEntrySetID
) -> EngineCacheEntrySet:
"""Load a EngineCacheEntrySet from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadEngineCacheEntrySetFromID", _args)
return EngineCacheEntrySet(_ctx)
[docs]
def load_engine_cache_from_id(self, id: EngineCacheID) -> EngineCache:
"""Load a EngineCache from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadEngineCacheFromID", _args)
return EngineCache(_ctx)
[docs]
def load_engine_from_id(self, id: EngineID) -> Engine:
"""Load a Engine from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadEngineFromID", _args)
return Engine(_ctx)
[docs]
def load_enum_type_def_from_id(self, id: EnumTypeDefID) -> EnumTypeDef:
"""Load a EnumTypeDef from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadEnumTypeDefFromID", _args)
return EnumTypeDef(_ctx)
[docs]
def load_enum_value_type_def_from_id(
self, id: EnumValueTypeDefID
) -> EnumValueTypeDef:
"""Load a EnumValueTypeDef from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadEnumValueTypeDefFromID", _args)
return EnumValueTypeDef(_ctx)
[docs]
def load_env_variable_from_id(self, id: EnvVariableID) -> EnvVariable:
"""Load a EnvVariable from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadEnvVariableFromID", _args)
return EnvVariable(_ctx)
[docs]
def load_error_from_id(self, id: ErrorID) -> Error:
"""Load a Error from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadErrorFromID", _args)
return Error(_ctx)
[docs]
def load_field_type_def_from_id(self, id: FieldTypeDefID) -> FieldTypeDef:
"""Load a FieldTypeDef from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadFieldTypeDefFromID", _args)
return FieldTypeDef(_ctx)
[docs]
def load_file_from_id(self, id: FileID) -> File:
"""Load a File from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadFileFromID", _args)
return File(_ctx)
[docs]
def load_function_arg_from_id(self, id: FunctionArgID) -> FunctionArg:
"""Load a FunctionArg from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadFunctionArgFromID", _args)
return FunctionArg(_ctx)
[docs]
def load_function_call_arg_value_from_id(
self, id: FunctionCallArgValueID
) -> FunctionCallArgValue:
"""Load a FunctionCallArgValue from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadFunctionCallArgValueFromID", _args)
return FunctionCallArgValue(_ctx)
[docs]
def load_function_call_from_id(self, id: FunctionCallID) -> FunctionCall:
"""Load a FunctionCall from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadFunctionCallFromID", _args)
return FunctionCall(_ctx)
[docs]
def load_function_from_id(self, id: FunctionID) -> Function:
"""Load a Function from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadFunctionFromID", _args)
return Function(_ctx)
[docs]
def load_generated_code_from_id(self, id: GeneratedCodeID) -> GeneratedCode:
"""Load a GeneratedCode from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadGeneratedCodeFromID", _args)
return GeneratedCode(_ctx)
[docs]
def load_git_module_source_from_id(self, id: GitModuleSourceID) -> GitModuleSource:
"""Load a GitModuleSource from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadGitModuleSourceFromID", _args)
return GitModuleSource(_ctx)
[docs]
def load_git_ref_from_id(self, id: GitRefID) -> GitRef:
"""Load a GitRef from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadGitRefFromID", _args)
return GitRef(_ctx)
[docs]
def load_git_repository_from_id(self, id: GitRepositoryID) -> GitRepository:
"""Load a GitRepository from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadGitRepositoryFromID", _args)
return GitRepository(_ctx)
[docs]
def load_host_from_id(self, id: HostID) -> Host:
"""Load a Host from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadHostFromID", _args)
return Host(_ctx)
[docs]
def load_interface_type_def_from_id(
self, id: InterfaceTypeDefID
) -> InterfaceTypeDef:
"""Load a InterfaceTypeDef from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadInterfaceTypeDefFromID", _args)
return InterfaceTypeDef(_ctx)
[docs]
def load_label_from_id(self, id: LabelID) -> Label:
"""Load a Label from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadLabelFromID", _args)
return Label(_ctx)
[docs]
def load_list_type_def_from_id(self, id: ListTypeDefID) -> ListTypeDef:
"""Load a ListTypeDef from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadListTypeDefFromID", _args)
return ListTypeDef(_ctx)
[docs]
def load_local_module_source_from_id(
self, id: LocalModuleSourceID
) -> LocalModuleSource:
"""Load a LocalModuleSource from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadLocalModuleSourceFromID", _args)
return LocalModuleSource(_ctx)
[docs]
def load_module_dependency_from_id(
self, id: ModuleDependencyID
) -> ModuleDependency:
"""Load a ModuleDependency from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadModuleDependencyFromID", _args)
return ModuleDependency(_ctx)
[docs]
def load_module_from_id(self, id: ModuleID) -> Module:
"""Load a Module from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadModuleFromID", _args)
return Module(_ctx)
[docs]
def load_module_source_from_id(self, id: ModuleSourceID) -> ModuleSource:
"""Load a ModuleSource from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadModuleSourceFromID", _args)
return ModuleSource(_ctx)
[docs]
def load_module_source_view_from_id(
self, id: ModuleSourceViewID
) -> ModuleSourceView:
"""Load a ModuleSourceView from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadModuleSourceViewFromID", _args)
return ModuleSourceView(_ctx)
[docs]
def load_object_type_def_from_id(self, id: ObjectTypeDefID) -> ObjectTypeDef:
"""Load a ObjectTypeDef from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadObjectTypeDefFromID", _args)
return ObjectTypeDef(_ctx)
[docs]
def load_port_from_id(self, id: PortID) -> Port:
"""Load a Port from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadPortFromID", _args)
return Port(_ctx)
[docs]
def load_scalar_type_def_from_id(self, id: ScalarTypeDefID) -> "ScalarTypeDef":
"""Load a ScalarTypeDef from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadScalarTypeDefFromID", _args)
return ScalarTypeDef(_ctx)
[docs]
def load_secret_from_id(self, id: SecretID) -> "Secret":
"""Load a Secret from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadSecretFromID", _args)
return Secret(_ctx)
[docs]
def load_service_from_id(self, id: ServiceID) -> "Service":
"""Load a Service from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadServiceFromID", _args)
return Service(_ctx)
[docs]
def load_socket_from_id(self, id: SocketID) -> "Socket":
"""Load a Socket from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadSocketFromID", _args)
return Socket(_ctx)
[docs]
def load_source_map_from_id(self, id: SourceMapID) -> "SourceMap":
"""Load a SourceMap from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadSourceMapFromID", _args)
return SourceMap(_ctx)
[docs]
def load_terminal_from_id(self, id: TerminalID) -> "Terminal":
"""Load a Terminal from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadTerminalFromID", _args)
return Terminal(_ctx)
[docs]
def load_type_def_from_id(self, id: TypeDefID) -> "TypeDef":
"""Load a TypeDef from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("loadTypeDefFromID", _args)
return TypeDef(_ctx)
[docs]
def module(self) -> Module:
"""Create a new module."""
_args: list[Arg] = []
_ctx = self._select("module", _args)
return Module(_ctx)
[docs]
def module_dependency(
self,
source: ModuleSource,
*,
name: str | None = "",
) -> ModuleDependency:
"""Create a new module dependency configuration from a module source and
name
Parameters
----------
source:
The source of the dependency
name:
If set, the name to use for the dependency. Otherwise, once
installed to a parent module, the name of the dependency module
will be used by default.
"""
_args = [
Arg("source", source),
Arg("name", name, ""),
]
_ctx = self._select("moduleDependency", _args)
return ModuleDependency(_ctx)
[docs]
def module_source(
self,
ref_string: str,
*,
ref_pin: str | None = "",
stable: bool | None = False,
rel_host_path: str | None = "",
) -> ModuleSource:
"""Create a new module source instance from a source ref string.
Parameters
----------
ref_string:
The string ref representation of the module source
ref_pin:
The pinned version of the module source
stable:
If true, enforce that the source is a stable version for source
kinds that support versioning.
rel_host_path:
The relative path to the module root from the host directory
"""
_args = [
Arg("refString", ref_string),
Arg("refPin", ref_pin, ""),
Arg("stable", stable, False),
Arg("relHostPath", rel_host_path, ""),
]
_ctx = self._select("moduleSource", _args)
return ModuleSource(_ctx)
[docs]
def secret(
self,
name: str,
*,
accessor: str | None = None,
) -> "Secret":
"""Reference a secret by name."""
_args = [
Arg("name", name),
Arg("accessor", accessor, None),
]
_ctx = self._select("secret", _args)
return Secret(_ctx)
[docs]
def set_secret(self, name: str, plaintext: str) -> "Secret":
"""Sets a secret given a user defined name to its plaintext and returns
the secret.
The plaintext value is limited to a size of 128000 bytes.
Parameters
----------
name:
The user defined name for this secret
plaintext:
The plaintext of the secret
"""
_args = [
Arg("name", name),
Arg("plaintext", plaintext),
]
_ctx = self._select("setSecret", _args)
return Secret(_ctx)
[docs]
def source_map(
self,
filename: str,
line: int,
column: int,
) -> "SourceMap":
"""Creates source map metadata.
Parameters
----------
filename:
The filename from the module source.
line:
The line number within the filename.
column:
The column number within the line.
"""
_args = [
Arg("filename", filename),
Arg("line", line),
Arg("column", column),
]
_ctx = self._select("sourceMap", _args)
return SourceMap(_ctx)
[docs]
def type_def(self) -> "TypeDef":
"""Create a new TypeDef."""
_args: list[Arg] = []
_ctx = self._select("typeDef", _args)
return TypeDef(_ctx)
[docs]
async def version(self) -> str:
"""Get the current Dagger Engine version.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("version", _args)
return await _ctx.execute(str)
@typecheck
class ScalarTypeDef(Type):
"""A definition of a custom scalar defined in a Module."""
[docs]
async def description(self) -> str:
"""A doc string for the scalar, if any.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("description", _args)
return await _ctx.execute(str)
[docs]
async def id(self) -> ScalarTypeDefID:
"""A unique identifier for this ScalarTypeDef.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
ScalarTypeDefID
The `ScalarTypeDefID` scalar type represents an identifier for an
object of type ScalarTypeDef.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(ScalarTypeDefID)
[docs]
async def name(self) -> str:
"""The name of the scalar.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("name", _args)
return await _ctx.execute(str)
[docs]
async def source_module_name(self) -> str:
"""If this ScalarTypeDef is associated with a Module, the name of the
module. Unset otherwise.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("sourceModuleName", _args)
return await _ctx.execute(str)
@typecheck
class Secret(Type):
"""A reference to a secret value, which can be handled more safely
than the value itself."""
[docs]
async def id(self) -> SecretID:
"""A unique identifier for this Secret.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
SecretID
The `SecretID` scalar type represents an identifier for an object
of type Secret.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(SecretID)
[docs]
async def name(self) -> str:
"""The name of this secret.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("name", _args)
return await _ctx.execute(str)
[docs]
async def plaintext(self) -> str:
"""The value of this secret.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("plaintext", _args)
return await _ctx.execute(str)
@typecheck
class Service(Type):
"""A content-addressed service providing TCP connectivity."""
[docs]
async def endpoint(
self,
*,
port: int | None = None,
scheme: str | None = "",
) -> str:
"""Retrieves an endpoint that clients can use to reach this container.
If no port is specified, the first exposed port is used. If none exist
an error is returned.
If a scheme is specified, a URL is returned. Otherwise, a host:port
pair is returned.
Parameters
----------
port:
The exposed port number for the endpoint
scheme:
Return a URL with the given scheme, eg. http for http://
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args = [
Arg("port", port, None),
Arg("scheme", scheme, ""),
]
_ctx = self._select("endpoint", _args)
return await _ctx.execute(str)
[docs]
async def hostname(self) -> str:
"""Retrieves a hostname which can be used by clients to reach this
container.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("hostname", _args)
return await _ctx.execute(str)
[docs]
async def id(self) -> ServiceID:
"""A unique identifier for this Service.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
ServiceID
The `ServiceID` scalar type represents an identifier for an object
of type Service.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(ServiceID)
[docs]
async def ports(self) -> list[Port]:
"""Retrieves the list of ports provided by the service."""
_args: list[Arg] = []
_ctx = self._select("ports", _args)
_ctx = Port(_ctx)._select("id", [])
@dataclass
class Response:
id: PortID
_ids = await _ctx.execute(list[Response])
return [
Port(
Client.from_context(_ctx)._select(
"loadPortFromID",
[Arg("id", v.id)],
)
)
for v in _ids
]
[docs]
async def start(self) -> Self:
"""Start the service and wait for its health checks to succeed.
Services bound to a Container do not need to be manually started.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("start", _args)
_id = await _ctx.execute(ServiceID)
_ctx = Client.from_context(_ctx)._select("loadServiceFromID", [Arg("id", _id)])
return Service(_ctx)
[docs]
async def stop(self, *, kill: bool | None = False) -> Self:
"""Stop the service.
Parameters
----------
kill:
Immediately kill the service without waiting for a graceful exit
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args = [
Arg("kill", kill, False),
]
_ctx = self._select("stop", _args)
_id = await _ctx.execute(ServiceID)
_ctx = Client.from_context(_ctx)._select("loadServiceFromID", [Arg("id", _id)])
return Service(_ctx)
[docs]
async def up(
self,
*,
ports: list[PortForward] | None = None,
random: bool | None = False,
) -> Void | None:
"""Creates a tunnel that forwards traffic from the caller's network to
this service.
Parameters
----------
ports:
List of frontend/backend port mappings to forward.
Frontend is the port accepting traffic on the host, backend is the
service port.
random:
Bind each tunnel port to a random port on the host.
Returns
-------
Void | None
The absence of a value. A Null Void is used as a placeholder for
resolvers that do not return anything.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args = [
Arg("ports", () if ports is None else ports, ()),
Arg("random", random, False),
]
_ctx = self._select("up", _args)
await _ctx.execute()
[docs]
def with_hostname(self, hostname: str) -> Self:
"""Configures a hostname which can be used by clients within the session
to reach this container.
Parameters
----------
hostname:
The hostname to use.
"""
_args = [
Arg("hostname", hostname),
]
_ctx = self._select("withHostname", _args)
return Service(_ctx)
[docs]
def with_(self, cb: Callable[["Service"], "Service"]) -> "Service":
"""Call the provided callable with current Service.
This is useful for reusability and readability by not breaking the calling chain.
"""
return cb(self)
@typecheck
class Socket(Type):
"""A Unix or TCP/IP socket that can be mounted into a container."""
[docs]
async def id(self) -> SocketID:
"""A unique identifier for this Socket.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
SocketID
The `SocketID` scalar type represents an identifier for an object
of type Socket.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(SocketID)
@typecheck
class SourceMap(Type):
"""Source location information."""
[docs]
async def column(self) -> int:
"""The column number within the line.
Returns
-------
int
The `Int` scalar type represents non-fractional signed whole
numeric values. Int can represent values between -(2^31) and 2^31
- 1.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("column", _args)
return await _ctx.execute(int)
[docs]
async def filename(self) -> str:
"""The filename from the module source.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("filename", _args)
return await _ctx.execute(str)
[docs]
async def id(self) -> SourceMapID:
"""A unique identifier for this SourceMap.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
SourceMapID
The `SourceMapID` scalar type represents an identifier for an
object of type SourceMap.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(SourceMapID)
[docs]
async def line(self) -> int:
"""The line number within the filename.
Returns
-------
int
The `Int` scalar type represents non-fractional signed whole
numeric values. Int can represent values between -(2^31) and 2^31
- 1.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("line", _args)
return await _ctx.execute(int)
[docs]
async def module(self) -> str:
"""The module dependency this was declared in.
Returns
-------
str
The `String` scalar type represents textual data, represented as
UTF-8 character sequences. The String type is most often used by
GraphQL to represent free-form human-readable text.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("module", _args)
return await _ctx.execute(str)
@typecheck
class Terminal(Type):
"""An interactive terminal that clients can connect to."""
[docs]
async def id(self) -> TerminalID:
"""A unique identifier for this Terminal.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
TerminalID
The `TerminalID` scalar type represents an identifier for an
object of type Terminal.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(TerminalID)
[docs]
async def sync(self) -> Self:
"""Forces evaluation of the pipeline in the engine.
It doesn't run the default command if no exec has been set.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("sync", _args)
_id = await _ctx.execute(TerminalID)
_ctx = Client.from_context(_ctx)._select("loadTerminalFromID", [Arg("id", _id)])
return Terminal(_ctx)
def __await__(self):
return self.sync().__await__()
@typecheck
class TypeDef(Type):
"""A definition of a parameter or return type in a Module."""
[docs]
def as_enum(self) -> EnumTypeDef:
"""If kind is ENUM, the enum-specific type definition. If kind is not
ENUM, this will be null.
"""
_args: list[Arg] = []
_ctx = self._select("asEnum", _args)
return EnumTypeDef(_ctx)
[docs]
def as_interface(self) -> InterfaceTypeDef:
"""If kind is INTERFACE, the interface-specific type definition. If kind
is not INTERFACE, this will be null.
"""
_args: list[Arg] = []
_ctx = self._select("asInterface", _args)
return InterfaceTypeDef(_ctx)
[docs]
def as_list(self) -> ListTypeDef:
"""If kind is LIST, the list-specific type definition. If kind is not
LIST, this will be null.
"""
_args: list[Arg] = []
_ctx = self._select("asList", _args)
return ListTypeDef(_ctx)
[docs]
def as_object(self) -> ObjectTypeDef:
"""If kind is OBJECT, the object-specific type definition. If kind is not
OBJECT, this will be null.
"""
_args: list[Arg] = []
_ctx = self._select("asObject", _args)
return ObjectTypeDef(_ctx)
[docs]
def as_scalar(self) -> ScalarTypeDef:
"""If kind is SCALAR, the scalar-specific type definition. If kind is not
SCALAR, this will be null.
"""
_args: list[Arg] = []
_ctx = self._select("asScalar", _args)
return ScalarTypeDef(_ctx)
[docs]
async def id(self) -> TypeDefID:
"""A unique identifier for this TypeDef.
Note
----
This is lazily evaluated, no operation is actually run.
Returns
-------
TypeDefID
The `TypeDefID` scalar type represents an identifier for an object
of type TypeDef.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("id", _args)
return await _ctx.execute(TypeDefID)
[docs]
async def kind(self) -> TypeDefKind:
"""The kind of type this is (e.g. primitive, list, object).
Returns
-------
TypeDefKind
Distinguishes the different kinds of TypeDefs.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("kind", _args)
return await _ctx.execute(TypeDefKind)
[docs]
async def optional(self) -> bool:
"""Whether this type can be set to null. Defaults to false.
Returns
-------
bool
The `Boolean` scalar type represents `true` or `false`.
Raises
------
ExecuteTimeoutError
If the time to execute the query exceeds the configured timeout.
QueryError
If the API returns an error.
"""
_args: list[Arg] = []
_ctx = self._select("optional", _args)
return await _ctx.execute(bool)
[docs]
def with_constructor(self, function: Function) -> Self:
"""Adds a function for constructing a new instance of an Object TypeDef,
failing if the type is not an object.
"""
_args = [
Arg("function", function),
]
_ctx = self._select("withConstructor", _args)
return TypeDef(_ctx)
[docs]
def with_enum(
self,
name: str,
*,
description: str | None = "",
source_map: SourceMap | None = None,
) -> Self:
"""Returns a TypeDef of kind Enum with the provided name.
Note that an enum's values may be omitted if the intent is only to
refer to an enum. This is how functions are able to return their own,
or any other circular reference.
Parameters
----------
name:
The name of the enum
description:
A doc string for the enum, if any
source_map:
The source map for the enum definition.
"""
_args = [
Arg("name", name),
Arg("description", description, ""),
Arg("sourceMap", source_map, None),
]
_ctx = self._select("withEnum", _args)
return TypeDef(_ctx)
[docs]
def with_enum_value(
self,
value: str,
*,
description: str | None = "",
source_map: SourceMap | None = None,
) -> Self:
"""Adds a static value for an Enum TypeDef, failing if the type is not an
enum.
Parameters
----------
value:
The name of the value in the enum
description:
A doc string for the value, if any
source_map:
The source map for the enum value definition.
"""
_args = [
Arg("value", value),
Arg("description", description, ""),
Arg("sourceMap", source_map, None),
]
_ctx = self._select("withEnumValue", _args)
return TypeDef(_ctx)
[docs]
def with_field(
self,
name: str,
type_def: Self,
*,
description: str | None = "",
source_map: SourceMap | None = None,
) -> Self:
"""Adds a static field for an Object TypeDef, failing if the type is not
an object.
Parameters
----------
name:
The name of the field in the object
type_def:
The type of the field
description:
A doc string for the field, if any
source_map:
The source map for the field definition.
"""
_args = [
Arg("name", name),
Arg("typeDef", type_def),
Arg("description", description, ""),
Arg("sourceMap", source_map, None),
]
_ctx = self._select("withField", _args)
return TypeDef(_ctx)
[docs]
def with_function(self, function: Function) -> Self:
"""Adds a function for an Object or Interface TypeDef, failing if the
type is not one of those kinds.
"""
_args = [
Arg("function", function),
]
_ctx = self._select("withFunction", _args)
return TypeDef(_ctx)
[docs]
def with_interface(
self,
name: str,
*,
description: str | None = "",
source_map: SourceMap | None = None,
) -> Self:
"""Returns a TypeDef of kind Interface with the provided name."""
_args = [
Arg("name", name),
Arg("description", description, ""),
Arg("sourceMap", source_map, None),
]
_ctx = self._select("withInterface", _args)
return TypeDef(_ctx)
[docs]
def with_kind(self, kind: TypeDefKind) -> Self:
"""Sets the kind of the type."""
_args = [
Arg("kind", kind),
]
_ctx = self._select("withKind", _args)
return TypeDef(_ctx)
[docs]
def with_list_of(self, element_type: Self) -> Self:
"""Returns a TypeDef of kind List with the provided type for its
elements.
"""
_args = [
Arg("elementType", element_type),
]
_ctx = self._select("withListOf", _args)
return TypeDef(_ctx)
[docs]
def with_object(
self,
name: str,
*,
description: str | None = "",
source_map: SourceMap | None = None,
) -> Self:
"""Returns a TypeDef of kind Object with the provided name.
Note that an object's fields and functions may be omitted if the
intent is only to refer to an object. This is how functions are able
to return their own object, or any other circular reference.
"""
_args = [
Arg("name", name),
Arg("description", description, ""),
Arg("sourceMap", source_map, None),
]
_ctx = self._select("withObject", _args)
return TypeDef(_ctx)
[docs]
def with_optional(self, optional: bool) -> Self:
"""Sets whether this type can be set to null."""
_args = [
Arg("optional", optional),
]
_ctx = self._select("withOptional", _args)
return TypeDef(_ctx)
[docs]
def with_scalar(
self,
name: str,
*,
description: str | None = "",
) -> Self:
"""Returns a TypeDef of kind Scalar with the provided name."""
_args = [
Arg("name", name),
Arg("description", description, ""),
]
_ctx = self._select("withScalar", _args)
return TypeDef(_ctx)
[docs]
def with_(self, cb: Callable[["TypeDef"], "TypeDef"]) -> "TypeDef":
"""Call the provided callable with current TypeDef.
This is useful for reusability and readability by not breaking the calling chain.
"""
return cb(self)
dag = Client()
"""The global client instance."""
__all__ = [
"JSON",
"BuildArg",
"CacheSharingMode",
"CacheVolume",
"CacheVolumeID",
"Client",
"Container",
"ContainerID",
"CurrentModule",
"CurrentModuleID",
"Directory",
"DirectoryID",
"Engine",
"EngineCache",
"EngineCacheEntry",
"EngineCacheEntryID",
"EngineCacheEntrySet",
"EngineCacheEntrySetID",
"EngineCacheID",
"EngineID",
"EnumTypeDef",
"EnumTypeDefID",
"EnumValueTypeDef",
"EnumValueTypeDefID",
"EnvVariable",
"EnvVariableID",
"Error",
"ErrorID",
"FieldTypeDef",
"FieldTypeDefID",
"File",
"FileID",
"Function",
"FunctionArg",
"FunctionArgID",
"FunctionCall",
"FunctionCallArgValue",
"FunctionCallArgValueID",
"FunctionCallID",
"FunctionID",
"GeneratedCode",
"GeneratedCodeID",
"GitModuleSource",
"GitModuleSourceID",
"GitRef",
"GitRefID",
"GitRepository",
"GitRepositoryID",
"Host",
"HostID",
"ImageLayerCompression",
"ImageMediaTypes",
"InputTypeDef",
"InputTypeDefID",
"InterfaceTypeDef",
"InterfaceTypeDefID",
"Label",
"LabelID",
"ListTypeDef",
"ListTypeDefID",
"LocalModuleSource",
"LocalModuleSourceID",
"Module",
"ModuleDependency",
"ModuleDependencyID",
"ModuleID",
"ModuleSource",
"ModuleSourceID",
"ModuleSourceKind",
"ModuleSourceView",
"ModuleSourceViewID",
"NetworkProtocol",
"ObjectTypeDef",
"ObjectTypeDefID",
"PipelineLabel",
"Platform",
"Port",
"PortForward",
"PortID",
"ReturnType",
"ScalarTypeDef",
"ScalarTypeDefID",
"Secret",
"SecretID",
"Service",
"ServiceID",
"Socket",
"SocketID",
"SourceMap",
"SourceMapID",
"Terminal",
"TerminalID",
"TypeDef",
"TypeDefID",
"TypeDefKind",
"Void",
"dag",
]