# Code generated by dagger. DO NOT EDIT.
import warnings # noqa: F401
from collections.abc import Callable, Sequence
from dataclasses import dataclass
from typing import Optional
from ._core import Arg, Root
from ._guards import typecheck
from .base import Enum, Input, Scalar, Type
class CacheID(Scalar):
"""A global cache volume identifier."""
class ContainerID(Scalar):
"""A unique container identifier. Null designates an empty container
(scratch)."""
class DirectoryID(Scalar):
"""A content-addressed directory identifier."""
class FileID(Scalar):
"""A file identifier."""
class FunctionID(Scalar):
"""A reference to a Function."""
class GeneratedCodeID(Scalar):
"""A reference to GeneratedCode."""
class JSON(Scalar):
"""An arbitrary JSON-encoded value."""
class ModuleID(Scalar):
"""A reference to a Module."""
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 SecretID(Scalar):
"""A unique identifier for a secret."""
class SocketID(Scalar):
"""A content-addressed socket identifier."""
class TypeDefID(Scalar):
"""A reference to a TypeDef."""
class Void(Scalar):
"""The absense 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 NetworkProtocol(Enum):
"""Transport layer network protocol associated to a port."""
TCP = "TCP"
"""TCP (Transmission Control Protocol)"""
UDP = "UDP"
"""UDP (User Datagram Protocol)"""
class TypeDefKind(Enum):
"""Distinguishes the different kinds of TypeDefs."""
BooleanKind = "BooleanKind"
"""A boolean value"""
IntegerKind = "IntegerKind"
"""An integer value"""
ListKind = "ListKind"
"""A list of values all having the same type.
Always paired with a ListTypeDef.
"""
ObjectKind = "ObjectKind"
"""A named type defined in the GraphQL schema, with fields and functions.
Always paired with an ObjectTypeDef.
"""
StringKind = "StringKind"
"""A string value"""
VoidKind = "VoidKind"
"""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.
"""
@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."""
@dataclass(slots=True)
class FunctionCallInput(Input):
name: str
"""The name of the argument to the function"""
value: JSON
"""The value of the argument represented as a string of the JSON serialization."""
@dataclass(slots=True)
class ModuleEnvironmentVariable(Input):
name: str
value: Optional[str] = None
@dataclass(slots=True)
class PipelineLabel(Input):
"""Key value object that represents a Pipeline label."""
name: str
"""Label name."""
value: str
"""Label value."""
class CacheVolume(Type):
"""A directory whose contents persist across runs."""
[docs]
@typecheck
async def id(self) -> CacheID:
"""Note
----
This is lazyly evaluated, no operation is actually run.
Returns
-------
CacheID
A global cache volume identifier.
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(CacheID)
@classmethod
def _id_type(cls) -> type[Scalar]:
return CacheID
class Container(Type):
"""An OCI-compatible container, also known as a docker container."""
[docs]
@typecheck
def build(
self,
context: "Directory",
*,
dockerfile: Optional[str] = None,
build_args: Optional[Sequence[BuildArg]] = None,
target: Optional[str] = None,
secrets: Optional[Sequence["Secret"]] = None,
) -> "Container":
"""Initializes this container from a Dockerfile build.
Parameters
----------
context:
Directory context used by the Dockerfile.
dockerfile:
Path to the Dockerfile to use.
Default: './Dockerfile'.
build_args:
Additional build arguments.
target:
Target build stage to build.
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 url?token=$(cat
/run/secrets/my-secret)"
"""
_args = [
Arg("context", context),
Arg("dockerfile", dockerfile, None),
Arg("buildArgs", build_args, None),
Arg("target", target, None),
Arg("secrets", secrets, None),
]
_ctx = self._select("build", _args)
return Container(_ctx)
[docs]
@typecheck
async def default_args(self) -> Optional[list[str]]:
"""Retrieves default arguments for future commands.
Returns
-------
Optional[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(Optional[list[str]])
[docs]
@typecheck
def directory(self, path: str) -> "Directory":
"""Retrieves a directory at the given path.
Mounts are included.
Parameters
----------
path:
The path of the directory to retrieve (e.g., "./src").
"""
_args = [
Arg("path", path),
]
_ctx = self._select("directory", _args)
return Directory(_ctx)
[docs]
@typecheck
async def endpoint(
self,
*,
port: Optional[int] = None,
scheme: Optional[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.
Currently experimental; set _EXPERIMENTAL_DAGGER_SERVICES_DNS=0 to
disable.
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, None),
]
_ctx = self._select("endpoint", _args)
return await _ctx.execute(str)
[docs]
@typecheck
async def entrypoint(self) -> Optional[list[str]]:
"""Retrieves entrypoint to be prepended to the arguments of all commands.
Returns
-------
Optional[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(Optional[list[str]])
[docs]
@typecheck
async def env_variable(self, name: str) -> Optional[str]:
"""Retrieves the value of the specified environment variable.
Parameters
----------
name:
The name of the environment variable to retrieve (e.g., "PATH").
Returns
-------
Optional[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("name", name),
]
_ctx = self._select("envVariable", _args)
return await _ctx.execute(Optional[str])
[docs]
@typecheck
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_multiple(
_name="name",
_value="value",
)
return await _ctx.execute(list[EnvVariable])
[docs]
@typecheck
async def export(
self,
path: str,
*,
platform_variants: Optional[Sequence["Container"]] = None,
forced_compression: Optional[ImageLayerCompression] = None,
media_types: Optional[ImageMediaTypes] = None,
) -> bool:
"""Writes the container as an OCI tarball to the destination file path on
the host for the specified platform variants.
Return true on success.
It can also publishes 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.
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 = [
Arg("path", path),
Arg("platformVariants", platform_variants, None),
Arg("forcedCompression", forced_compression, None),
Arg("mediaTypes", media_types, None),
]
_ctx = self._select("export", _args)
return await _ctx.execute(bool)
[docs]
@typecheck
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.
Currently experimental; set _EXPERIMENTAL_DAGGER_SERVICES_DNS=0 to
disable.
"""
_args: list[Arg] = []
_ctx = self._select("exposedPorts", _args)
_ctx = Port(_ctx)._select_multiple(
_description="description",
_port="port",
_protocol="protocol",
)
return await _ctx.execute(list[Port])
[docs]
@typecheck
def file(self, path: str) -> "File":
"""Retrieves a file at the given path.
Mounts are included.
Parameters
----------
path:
The path of the file to retrieve (e.g., "./README.md").
"""
_args = [
Arg("path", path),
]
_ctx = self._select("file", _args)
return File(_ctx)
[docs]
@typecheck
def from_(self, address: str) -> "Container":
"""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]
@typecheck
async def hostname(self) -> str:
"""Retrieves a hostname which can be used by clients to reach this
container.
Currently experimental; set _EXPERIMENTAL_DAGGER_SERVICES_DNS=0 to
disable.
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]
@typecheck
async def id(self) -> ContainerID:
"""A unique identifier for this container.
Note
----
This is lazyly evaluated, no operation is actually run.
Returns
-------
ContainerID
A unique container identifier. Null designates an empty container
(scratch).
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)
@classmethod
def _id_type(cls) -> type[Scalar]:
return ContainerID
@classmethod
def _from_id_query_field(cls):
return "container"
[docs]
@typecheck
async def image_ref(self) -> Optional[str]:
"""The unique image reference which can only be retrieved immediately
after the 'Container.From' call.
Returns
-------
Optional[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(Optional[str])
[docs]
@typecheck
def import_(
self,
source: "File",
*,
tag: Optional[str] = None,
) -> "Container":
"""Reads the container from an OCI tarball.
NOTE: this involves unpacking the tarball to an OCI store on the host
at
$XDG_CACHE_DIR/dagger/oci. This directory can be removed whenever you
like.
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, None),
]
_ctx = self._select("import", _args)
return Container(_ctx)
[docs]
@typecheck
async def label(self, name: str) -> Optional[str]:
"""Retrieves the value of the specified label.
Returns
-------
Optional[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("name", name),
]
_ctx = self._select("label", _args)
return await _ctx.execute(Optional[str])
[docs]
@typecheck
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_multiple(
_name="name",
_value="value",
)
return await _ctx.execute(list[Label])
[docs]
@typecheck
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]
@typecheck
def pipeline(
self,
name: str,
*,
description: Optional[str] = None,
labels: Optional[Sequence[PipelineLabel]] = None,
) -> "Container":
"""Creates a named sub-pipeline
Parameters
----------
name:
Pipeline name.
description:
Pipeline description.
labels:
Pipeline labels.
"""
_args = [
Arg("name", name),
Arg("description", description, None),
Arg("labels", labels, None),
]
_ctx = self._select("pipeline", _args)
return Container(_ctx)
[docs]
@typecheck
async def publish(
self,
address: str,
*,
platform_variants: Optional[Sequence["Container"]] = None,
forced_compression: Optional[ImageLayerCompression] = None,
media_types: Optional[ImageMediaTypes] = None,
) -> 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", platform_variants, None),
Arg("forcedCompression", forced_compression, None),
Arg("mediaTypes", media_types, None),
]
_ctx = self._select("publish", _args)
return await _ctx.execute(str)
[docs]
@typecheck
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]
@typecheck
async def shell_endpoint(self) -> str:
"""Return a websocket endpoint that, if connected to, will start the
container with a TTY streamed
over the websocket.
Primarily intended for internal use with the dagger CLI.
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("shellEndpoint", _args)
return await _ctx.execute(str)
[docs]
@typecheck
async def stderr(self) -> str:
"""The error stream of the last executed command.
Will execute default command if none is set, or error if there's no
default.
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]
@typecheck
async def stdout(self) -> str:
"""The output stream of the last executed command.
Will execute default command if none is set, or error if there's no
default.
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]
@typecheck
async def sync(self) -> "Container":
"""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("container", [Arg("id", _id)])
return Container(_ctx)
def __await__(self):
return self.sync().__await__()
[docs]
@typecheck
async def user(self) -> Optional[str]:
"""Retrieves the user to be set for all commands.
Returns
-------
Optional[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(Optional[str])
[docs]
@typecheck
def with_default_args(
self,
*,
args: Optional[Sequence[str]] = None,
) -> "Container":
"""Configures default arguments for future commands.
Parameters
----------
args:
Arguments to prepend to future executions (e.g., ["-v", "--no-
cache"]).
"""
_args = [
Arg("args", args, None),
]
_ctx = self._select("withDefaultArgs", _args)
return Container(_ctx)
[docs]
@typecheck
def with_directory(
self,
path: str,
directory: "Directory",
*,
exclude: Optional[Sequence[str]] = None,
include: Optional[Sequence[str]] = None,
owner: Optional[str] = None,
) -> "Container":
"""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.
"""
_args = [
Arg("path", path),
Arg("directory", directory),
Arg("exclude", exclude, None),
Arg("include", include, None),
Arg("owner", owner, None),
]
_ctx = self._select("withDirectory", _args)
return Container(_ctx)
[docs]
@typecheck
def with_entrypoint(self, args: Sequence[str]) -> "Container":
"""Retrieves this container but with a different command entrypoint.
Parameters
----------
args:
Entrypoint to use for future executions (e.g., ["go", "run"]).
"""
_args = [
Arg("args", args),
]
_ctx = self._select("withEntrypoint", _args)
return Container(_ctx)
[docs]
@typecheck
def with_env_variable(
self,
name: str,
value: str,
*,
expand: Optional[bool] = None,
) -> "Container":
"""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, None),
]
_ctx = self._select("withEnvVariable", _args)
return Container(_ctx)
[docs]
@typecheck
def with_exec(
self,
args: Sequence[str],
*,
skip_entrypoint: Optional[bool] = None,
stdin: Optional[str] = None,
redirect_stdout: Optional[str] = None,
redirect_stderr: Optional[str] = None,
experimental_privileged_nesting: Optional[bool] = None,
insecure_root_capabilities: Optional[bool] = None,
) -> "Container":
"""Retrieves this container after executing the specified command inside
it.
Parameters
----------
args:
Command to run instead of the container's default command (e.g.,
["run", "main.go"]).
If empty, the container's default command is used.
skip_entrypoint:
If the container has an entrypoint, ignore it for args rather than
using it to wrap them.
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").
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("skipEntrypoint", skip_entrypoint, None),
Arg("stdin", stdin, None),
Arg("redirectStdout", redirect_stdout, None),
Arg("redirectStderr", redirect_stderr, None),
Arg("experimentalPrivilegedNesting", experimental_privileged_nesting, None),
Arg("insecureRootCapabilities", insecure_root_capabilities, None),
]
_ctx = self._select("withExec", _args)
return Container(_ctx)
[docs]
@typecheck
def with_exposed_port(
self,
port: int,
*,
protocol: Optional[NetworkProtocol] = None,
description: Optional[str] = None,
) -> "Container":
"""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
Currently experimental; set _EXPERIMENTAL_DAGGER_SERVICES_DNS=0 to
disable.
Parameters
----------
port:
Port number to expose
protocol:
Transport layer network protocol
description:
Optional port description
"""
_args = [
Arg("port", port),
Arg("protocol", protocol, None),
Arg("description", description, None),
]
_ctx = self._select("withExposedPort", _args)
return Container(_ctx)
[docs]
@typecheck
def with_file(
self,
path: str,
source: "File",
*,
permissions: Optional[int] = None,
owner: Optional[str] = None,
) -> "Container":
"""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).
Default: 0644.
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.
"""
_args = [
Arg("path", path),
Arg("source", source),
Arg("permissions", permissions, None),
Arg("owner", owner, None),
]
_ctx = self._select("withFile", _args)
return Container(_ctx)
[docs]
@typecheck
def with_focus(self) -> "Container":
"""Indicate that subsequent operations should be featured more
prominently in
the UI.
"""
_args: list[Arg] = []
_ctx = self._select("withFocus", _args)
return Container(_ctx)
[docs]
@typecheck
def with_label(self, name: str, value: str) -> "Container":
"""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]
@typecheck
def with_mounted_cache(
self,
path: str,
cache: CacheVolume,
*,
source: Optional["Directory"] = None,
sharing: Optional[CacheSharingMode] = None,
owner: Optional[str] = None,
) -> "Container":
"""Retrieves this container plus a cache volume mounted at the given
path.
Parameters
----------
path:
Location of the cache directory (e.g., "/cache/node_modules").
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.
"""
_args = [
Arg("path", path),
Arg("cache", cache),
Arg("source", source, None),
Arg("sharing", sharing, None),
Arg("owner", owner, None),
]
_ctx = self._select("withMountedCache", _args)
return Container(_ctx)
[docs]
@typecheck
def with_mounted_directory(
self,
path: str,
source: "Directory",
*,
owner: Optional[str] = None,
) -> "Container":
"""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.
"""
_args = [
Arg("path", path),
Arg("source", source),
Arg("owner", owner, None),
]
_ctx = self._select("withMountedDirectory", _args)
return Container(_ctx)
[docs]
@typecheck
def with_mounted_file(
self,
path: str,
source: "File",
*,
owner: Optional[str] = None,
) -> "Container":
"""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.
"""
_args = [
Arg("path", path),
Arg("source", source),
Arg("owner", owner, None),
]
_ctx = self._select("withMountedFile", _args)
return Container(_ctx)
[docs]
@typecheck
def with_mounted_secret(
self,
path: str,
source: "Secret",
*,
owner: Optional[str] = None,
mode: Optional[int] = None,
) -> "Container":
"""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.
Default: 0400.
"""
_args = [
Arg("path", path),
Arg("source", source),
Arg("owner", owner, None),
Arg("mode", mode, None),
]
_ctx = self._select("withMountedSecret", _args)
return Container(_ctx)
[docs]
@typecheck
def with_mounted_temp(self, path: str) -> "Container":
"""Retrieves this container plus a temporary directory mounted at the
given path.
Parameters
----------
path:
Location of the temporary directory (e.g., "/tmp/temp_dir").
"""
_args = [
Arg("path", path),
]
_ctx = self._select("withMountedTemp", _args)
return Container(_ctx)
[docs]
@typecheck
def with_new_file(
self,
path: str,
*,
contents: Optional[str] = None,
permissions: Optional[int] = None,
owner: Optional[str] = None,
) -> "Container":
"""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).
Default: 0644.
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.
"""
_args = [
Arg("path", path),
Arg("contents", contents, None),
Arg("permissions", permissions, None),
Arg("owner", owner, None),
]
_ctx = self._select("withNewFile", _args)
return Container(_ctx)
[docs]
@typecheck
def with_registry_auth(
self,
address: str,
username: str,
secret: "Secret",
) -> "Container":
"""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]
@typecheck
def with_rootfs(self, directory: "Directory") -> "Container":
"""Initializes this container from this DirectoryID."""
_args = [
Arg("directory", directory),
]
_ctx = self._select("withRootfs", _args)
return Container(_ctx)
[docs]
@typecheck
def with_secret_variable(self, name: str, secret: "Secret") -> "Container":
"""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]
@typecheck
def with_service_binding(self, alias: str, service: "Container") -> "Container":
"""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.
Currently experimental; set _EXPERIMENTAL_DAGGER_SERVICES_DNS=0 to
disable.
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]
@typecheck
def with_unix_socket(
self,
path: str,
source: "Socket",
*,
owner: Optional[str] = None,
) -> "Container":
"""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.
"""
_args = [
Arg("path", path),
Arg("source", source),
Arg("owner", owner, None),
]
_ctx = self._select("withUnixSocket", _args)
return Container(_ctx)
[docs]
@typecheck
def with_user(self, name: str) -> "Container":
"""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]
@typecheck
def with_workdir(self, path: str) -> "Container":
"""Retrieves this container with a different working directory.
Parameters
----------
path:
The path to set as the working directory (e.g., "/app").
"""
_args = [
Arg("path", path),
]
_ctx = self._select("withWorkdir", _args)
return Container(_ctx)
[docs]
@typecheck
def without_env_variable(self, name: str) -> "Container":
"""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]
@typecheck
def without_exposed_port(
self,
port: int,
*,
protocol: Optional[NetworkProtocol] = None,
) -> "Container":
"""Unexpose a previously exposed port.
Currently experimental; set _EXPERIMENTAL_DAGGER_SERVICES_DNS=0 to
disable.
Parameters
----------
port:
Port number to unexpose
protocol:
Port protocol to unexpose
"""
_args = [
Arg("port", port),
Arg("protocol", protocol, None),
]
_ctx = self._select("withoutExposedPort", _args)
return Container(_ctx)
[docs]
@typecheck
def without_focus(self) -> "Container":
"""Indicate that subsequent operations should not be featured more
prominently
in the UI.
This is the initial state of all containers.
"""
_args: list[Arg] = []
_ctx = self._select("withoutFocus", _args)
return Container(_ctx)
[docs]
@typecheck
def without_label(self, name: str) -> "Container":
"""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]
@typecheck
def without_mount(self, path: str) -> "Container":
"""Retrieves this container after unmounting everything at the given
path.
Parameters
----------
path:
Location of the cache directory (e.g., "/cache/node_modules").
"""
_args = [
Arg("path", path),
]
_ctx = self._select("withoutMount", _args)
return Container(_ctx)
[docs]
@typecheck
def without_registry_auth(self, address: str) -> "Container":
"""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]
@typecheck
def without_unix_socket(self, path: str) -> "Container":
"""Retrieves this container with a previously added Unix socket removed.
Parameters
----------
path:
Location of the socket to remove (e.g., "/tmp/socket").
"""
_args = [
Arg("path", path),
]
_ctx = self._select("withoutUnixSocket", _args)
return Container(_ctx)
[docs]
@typecheck
async def workdir(self) -> Optional[str]:
"""Retrieves the working directory for all commands.
Returns
-------
Optional[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(Optional[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)
class Directory(Type):
"""A directory."""
[docs]
@typecheck
def as_module(
self,
*,
source_subpath: Optional[str] = None,
runtime: Optional[Container] = None,
) -> "Module":
"""Load the directory as a Dagger module
Parameters
----------
source_subpath:
An optional subpath of the directory which contains the module's
source
code.
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.
runtime:
A pre-built runtime container to use instead of building one from
the
source code. This is useful for bootstrapping.
You should ignore this unless you're building a Dagger SDK.
"""
_args = [
Arg("sourceSubpath", source_subpath, None),
Arg("runtime", runtime, None),
]
_ctx = self._select("asModule", _args)
return Module(_ctx)
[docs]
@typecheck
def diff(self, other: "Directory") -> "Directory":
"""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]
@typecheck
def directory(self, path: str) -> "Directory":
"""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]
@typecheck
def docker_build(
self,
*,
dockerfile: Optional[str] = None,
platform: Optional[Platform] = None,
build_args: Optional[Sequence[BuildArg]] = None,
target: Optional[str] = None,
secrets: Optional[Sequence["Secret"]] = None,
) -> Container:
"""Builds a new Docker container from this directory.
Parameters
----------
dockerfile:
Path to the Dockerfile to use (e.g., "frontend.Dockerfile").
Defaults: './Dockerfile'.
platform:
The platform to build.
build_args:
Build arguments to use in the build.
target:
Target build stage to build.
secrets:
Secrets to pass to the build.
They will be mounted at /run/secrets/[secret-name].
"""
_args = [
Arg("dockerfile", dockerfile, None),
Arg("platform", platform, None),
Arg("buildArgs", build_args, None),
Arg("target", target, None),
Arg("secrets", secrets, None),
]
_ctx = self._select("dockerBuild", _args)
return Container(_ctx)
[docs]
@typecheck
async def entries(self, *, path: Optional[str] = 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]
@typecheck
async def export(self, path: str) -> bool:
"""Writes the contents of the directory to a path on the host.
Parameters
----------
path:
Location of the copied directory (e.g., "logs/").
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 = [
Arg("path", path),
]
_ctx = self._select("export", _args)
return await _ctx.execute(bool)
[docs]
@typecheck
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]
@typecheck
async def id(self) -> DirectoryID:
"""The content-addressed identifier of the directory.
Note
----
This is lazyly evaluated, no operation is actually run.
Returns
-------
DirectoryID
A content-addressed directory identifier.
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)
@classmethod
def _id_type(cls) -> type[Scalar]:
return DirectoryID
@classmethod
def _from_id_query_field(cls):
return "directory"
[docs]
@typecheck
def pipeline(
self,
name: str,
*,
description: Optional[str] = None,
labels: Optional[Sequence[PipelineLabel]] = None,
) -> "Directory":
"""Creates a named sub-pipeline
Parameters
----------
name:
Pipeline name.
description:
Pipeline description.
labels:
Pipeline labels.
"""
_args = [
Arg("name", name),
Arg("description", description, None),
Arg("labels", labels, None),
]
_ctx = self._select("pipeline", _args)
return Directory(_ctx)
[docs]
@typecheck
async def sync(self) -> "Directory":
"""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("directory", [Arg("id", _id)])
return Directory(_ctx)
def __await__(self):
return self.sync().__await__()
[docs]
@typecheck
def with_directory(
self,
path: str,
directory: "Directory",
*,
exclude: Optional[Sequence[str]] = None,
include: Optional[Sequence[str]] = None,
) -> "Directory":
"""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", exclude, None),
Arg("include", include, None),
]
_ctx = self._select("withDirectory", _args)
return Directory(_ctx)
[docs]
@typecheck
def with_file(
self,
path: str,
source: "File",
*,
permissions: Optional[int] = None,
) -> "Directory":
"""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).
Default: 0644.
"""
_args = [
Arg("path", path),
Arg("source", source),
Arg("permissions", permissions, None),
]
_ctx = self._select("withFile", _args)
return Directory(_ctx)
[docs]
@typecheck
def with_new_directory(
self,
path: str,
*,
permissions: Optional[int] = None,
) -> "Directory":
"""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).
Default: 0755.
"""
_args = [
Arg("path", path),
Arg("permissions", permissions, None),
]
_ctx = self._select("withNewDirectory", _args)
return Directory(_ctx)
[docs]
@typecheck
def with_new_file(
self,
path: str,
contents: str,
*,
permissions: Optional[int] = None,
) -> "Directory":
"""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).
Default: 0644.
"""
_args = [
Arg("path", path),
Arg("contents", contents),
Arg("permissions", permissions, None),
]
_ctx = self._select("withNewFile", _args)
return Directory(_ctx)
[docs]
@typecheck
def with_timestamps(self, timestamp: int) -> "Directory":
"""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]
@typecheck
def without_directory(self, path: str) -> "Directory":
"""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]
@typecheck
def without_file(self, path: str) -> "Directory":
"""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 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)
class EnvVariable(Type):
"""A simple key value object that represents an environment
variable."""
__slots__ = (
"_name",
"_value",
)
_name: Optional[str]
_value: Optional[str]
[docs]
@typecheck
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.
"""
if hasattr(self, "_name"):
return self._name
_args: list[Arg] = []
_ctx = self._select("name", _args)
return await _ctx.execute(str)
[docs]
@typecheck
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.
"""
if hasattr(self, "_value"):
return self._value
_args: list[Arg] = []
_ctx = self._select("value", _args)
return await _ctx.execute(str)
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)."""
__slots__ = (
"_description",
"_name",
)
_description: Optional[str]
_name: Optional[str]
[docs]
@typecheck
async def description(self) -> Optional[str]:
"""A doc string for the field, if any
Returns
-------
Optional[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.
"""
if hasattr(self, "_description"):
return self._description
_args: list[Arg] = []
_ctx = self._select("description", _args)
return await _ctx.execute(Optional[str])
[docs]
@typecheck
async def name(self) -> str:
"""The name of the field in 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.
"""
if hasattr(self, "_name"):
return self._name
_args: list[Arg] = []
_ctx = self._select("name", _args)
return await _ctx.execute(str)
[docs]
@typecheck
def type_def(self) -> "TypeDef":
"""The type of the field"""
_args: list[Arg] = []
_ctx = self._select("typeDef", _args)
return TypeDef(_ctx)
class File(Type):
"""A file."""
[docs]
@typecheck
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]
@typecheck
async def export(
self,
path: str,
*,
allow_parent_dir_path: Optional[bool] = None,
) -> bool:
"""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
-------
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 = [
Arg("path", path),
Arg("allowParentDirPath", allow_parent_dir_path, None),
]
_ctx = self._select("export", _args)
return await _ctx.execute(bool)
[docs]
@typecheck
async def id(self) -> FileID:
"""Retrieves the content-addressed identifier of the file.
Note
----
This is lazyly evaluated, no operation is actually run.
Returns
-------
FileID
A file identifier.
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)
@classmethod
def _id_type(cls) -> type[Scalar]:
return FileID
@classmethod
def _from_id_query_field(cls):
return "file"
[docs]
@typecheck
async def size(self) -> int:
"""Gets 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]
@typecheck
async def sync(self) -> "File":
"""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("file", [Arg("id", _id)])
return File(_ctx)
def __await__(self):
return self.sync().__await__()
[docs]
@typecheck
def with_timestamps(self, timestamp: int) -> "File":
"""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)
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."""
__slots__ = (
"_call",
"_description",
"_name",
)
_call: Optional[JSON]
_description: Optional[str]
_name: Optional[str]
[docs]
@typecheck
async def args(self) -> list["FunctionArg"]:
"""Arguments accepted by this function, if any"""
_args: list[Arg] = []
_ctx = self._select("args", _args)
_ctx = FunctionArg(_ctx)._select_multiple(
_default_value="defaultValue",
_description="description",
_name="name",
)
return await _ctx.execute(list[FunctionArg])
[docs]
@typecheck
async def call(
self,
*,
input: Optional[Sequence[FunctionCallInput]] = None,
) -> JSON:
"""Execute this function using dynamic input+output types.
Typically, it's preferable to invoke a function using a type
safe graphql query rather than using this call field. However,
call is useful for some advanced use cases where dynamically
loading arbitrary modules and invoking functions in them is
required.
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.
"""
if hasattr(self, "_call"):
return self._call
_args = [
Arg("input", input, None),
]
_ctx = self._select("call", _args)
return await _ctx.execute(JSON)
[docs]
@typecheck
async def description(self) -> Optional[str]:
"""A doc string for the function, if any
Returns
-------
Optional[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.
"""
if hasattr(self, "_description"):
return self._description
_args: list[Arg] = []
_ctx = self._select("description", _args)
return await _ctx.execute(Optional[str])
[docs]
@typecheck
async def id(self) -> FunctionID:
"""The ID of the function
Note
----
This is lazyly evaluated, no operation is actually run.
Returns
-------
FunctionID
A reference to a 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)
@classmethod
def _id_type(cls) -> type[Scalar]:
return FunctionID
@classmethod
def _from_id_query_field(cls):
return "function"
[docs]
@typecheck
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.
"""
if hasattr(self, "_name"):
return self._name
_args: list[Arg] = []
_ctx = self._select("name", _args)
return await _ctx.execute(str)
[docs]
@typecheck
def return_type(self) -> "TypeDef":
"""The type returned by this function"""
_args: list[Arg] = []
_ctx = self._select("returnType", _args)
return TypeDef(_ctx)
[docs]
@typecheck
def with_arg(
self,
name: str,
type_def: "TypeDef",
*,
description: Optional[str] = None,
default_value: Optional[JSON] = None,
) -> "Function":
"""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
"""
_args = [
Arg("name", name),
Arg("typeDef", type_def),
Arg("description", description, None),
Arg("defaultValue", default_value, None),
]
_ctx = self._select("withArg", _args)
return Function(_ctx)
[docs]
@typecheck
def with_description(self, description: str) -> "Function":
"""Returns the function with the doc string"""
_args = [
Arg("description", description),
]
_ctx = self._select("withDescription", _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)
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."""
__slots__ = (
"_default_value",
"_description",
"_name",
)
_default_value: Optional[JSON]
_description: Optional[str]
_name: Optional[str]
[docs]
@typecheck
async def default_value(self) -> Optional[JSON]:
"""A default value to use for this argument when not explicitly set by
the caller, if any
Returns
-------
Optional[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.
"""
if hasattr(self, "_default_value"):
return self._default_value
_args: list[Arg] = []
_ctx = self._select("defaultValue", _args)
return await _ctx.execute(Optional[JSON])
[docs]
@typecheck
async def description(self) -> Optional[str]:
"""A doc string for the argument, if any
Returns
-------
Optional[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.
"""
if hasattr(self, "_description"):
return self._description
_args: list[Arg] = []
_ctx = self._select("description", _args)
return await _ctx.execute(Optional[str])
[docs]
@typecheck
async def name(self) -> Optional[str]:
"""The name of the argument
Returns
-------
Optional[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.
"""
if hasattr(self, "_name"):
return self._name
_args: list[Arg] = []
_ctx = self._select("name", _args)
return await _ctx.execute(Optional[str])
[docs]
@typecheck
def type_def(self) -> "TypeDef":
"""The type of the argument"""
_args: list[Arg] = []
_ctx = self._select("typeDef", _args)
return TypeDef(_ctx)
class FunctionCall(Type):
[docs]
@typecheck
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]
@typecheck
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]
@typecheck
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]
@typecheck
async def return_value(self, value: JSON) -> Optional[Void]:
"""Set the return value of the function call to the provided value.
The value should be a string of the JSON serialization of the return
value.
Returns
-------
Optional[Void]
The absense 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)
return await _ctx.execute(Optional[Void])
class FunctionCallArgValue(Type):
__slots__ = (
"_name",
"_value",
)
_name: Optional[str]
_value: Optional[JSON]
[docs]
@typecheck
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.
"""
if hasattr(self, "_name"):
return self._name
_args: list[Arg] = []
_ctx = self._select("name", _args)
return await _ctx.execute(str)
[docs]
@typecheck
async def value(self) -> JSON:
"""The value of the argument represented as a string of the JSON
serialization.
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.
"""
if hasattr(self, "_value"):
return self._value
_args: list[Arg] = []
_ctx = self._select("value", _args)
return await _ctx.execute(JSON)
class GeneratedCode(Type):
[docs]
@typecheck
def code(self) -> Directory:
"""The directory containing the generated code"""
_args: list[Arg] = []
_ctx = self._select("code", _args)
return Directory(_ctx)
[docs]
@typecheck
async def id(self) -> GeneratedCodeID:
"""Note
----
This is lazyly evaluated, no operation is actually run.
Returns
-------
GeneratedCodeID
A reference to 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)
@classmethod
def _id_type(cls) -> type[Scalar]:
return GeneratedCodeID
@classmethod
def _from_id_query_field(cls):
return "generatedCode"
[docs]
@typecheck
async def vcs_generated_paths(self) -> Optional[list[str]]:
"""List of paths to mark generated in version control (i.e.
.gitattributes)
Returns
-------
Optional[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(Optional[list[str]])
[docs]
@typecheck
async def vcs_ignored_paths(self) -> Optional[list[str]]:
"""List of paths to ignore in version control (i.e. .gitignore)
Returns
-------
Optional[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(Optional[list[str]])
[docs]
@typecheck
def with_code(self, code: Directory) -> "GeneratedCode":
"""Set the directory containing the generated code"""
_args = [
Arg("code", code),
]
_ctx = self._select("withCode", _args)
return GeneratedCode(_ctx)
[docs]
@typecheck
def with_vcs_generated_paths(self, paths: Sequence[str]) -> "GeneratedCode":
"""Set the list of paths to mark generated in version control"""
_args = [
Arg("paths", paths),
]
_ctx = self._select("withVCSGeneratedPaths", _args)
return GeneratedCode(_ctx)
[docs]
@typecheck
def with_vcs_ignored_paths(self, paths: Sequence[str]) -> "GeneratedCode":
"""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)
class GitRef(Type):
"""A git ref (tag, branch or commit)."""
[docs]
@typecheck
def tree(
self,
*,
ssh_known_hosts: Optional[str] = None,
ssh_auth_socket: Optional["Socket"] = None,
) -> Directory:
"""The filesystem tree at this ref."""
_args = [
Arg("sshKnownHosts", ssh_known_hosts, None),
Arg("sshAuthSocket", ssh_auth_socket, None),
]
_ctx = self._select("tree", _args)
return Directory(_ctx)
class GitRepository(Type):
"""A git repository."""
[docs]
@typecheck
def branch(self, name: str) -> GitRef:
"""Returns details on one branch.
Parameters
----------
name:
Branch's name (e.g., "main").
"""
_args = [
Arg("name", name),
]
_ctx = self._select("branch", _args)
return GitRef(_ctx)
[docs]
@typecheck
def commit(self, id: str) -> GitRef:
"""Returns details on one commit.
Parameters
----------
id:
Identifier of the commit (e.g.,
"b6315d8f2810962c601af73f86831f6866ea798b").
"""
_args = [
Arg("id", id),
]
_ctx = self._select("commit", _args)
return GitRef(_ctx)
[docs]
@typecheck
def tag(self, name: str) -> GitRef:
"""Returns details on one tag.
Parameters
----------
name:
Tag's name (e.g., "v0.3.9").
"""
_args = [
Arg("name", name),
]
_ctx = self._select("tag", _args)
return GitRef(_ctx)
class Host(Type):
"""Information about the host execution environment."""
[docs]
@typecheck
def directory(
self,
path: str,
*,
exclude: Optional[Sequence[str]] = None,
include: Optional[Sequence[str]] = 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", exclude, None),
Arg("include", include, None),
]
_ctx = self._select("directory", _args)
return Directory(_ctx)
[docs]
@typecheck
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]
@typecheck
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]
@typecheck
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)
class Label(Type):
"""A simple key value object that represents a label."""
__slots__ = (
"_name",
"_value",
)
_name: Optional[str]
_value: Optional[str]
[docs]
@typecheck
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.
"""
if hasattr(self, "_name"):
return self._name
_args: list[Arg] = []
_ctx = self._select("name", _args)
return await _ctx.execute(str)
[docs]
@typecheck
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.
"""
if hasattr(self, "_value"):
return self._value
_args: list[Arg] = []
_ctx = self._select("value", _args)
return await _ctx.execute(str)
class ListTypeDef(Type):
"""A definition of a list type in a Module."""
[docs]
@typecheck
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)
class Module(Type):
__slots__ = (
"_dependency_config",
"_description",
"_name",
"_sdk",
"_sdk_runtime",
"_serve",
"_source_directory_sub_path",
)
_dependency_config: Optional[str]
_description: Optional[str]
_name: Optional[str]
_sdk: Optional[str]
_sdk_runtime: Optional[str]
_serve: Optional[Void]
_source_directory_sub_path: Optional[str]
[docs]
@typecheck
async def dependencies(self) -> list["Module"]:
"""Modules used by this module"""
_args: list[Arg] = []
_ctx = self._select("dependencies", _args)
_ctx = Module(_ctx)._select_multiple(
_dependency_config="dependencyConfig",
_description="description",
_name="name",
_sdk="sdk",
_sdk_runtime="sdkRuntime",
_serve="serve",
_source_directory_sub_path="sourceDirectorySubPath",
)
return await _ctx.execute(list[Module])
[docs]
@typecheck
async def dependency_config(self) -> list[str]:
"""The dependencies as configured by the module
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.
"""
if hasattr(self, "_dependency_config"):
return self._dependency_config
_args: list[Arg] = []
_ctx = self._select("dependencyConfig", _args)
return await _ctx.execute(list[str])
[docs]
@typecheck
async def description(self) -> Optional[str]:
"""The doc string of the module, if any
Returns
-------
Optional[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.
"""
if hasattr(self, "_description"):
return self._description
_args: list[Arg] = []
_ctx = self._select("description", _args)
return await _ctx.execute(Optional[str])
[docs]
@typecheck
def generated_code(self) -> GeneratedCode:
"""The code generated by the SDK's runtime"""
_args: list[Arg] = []
_ctx = self._select("generatedCode", _args)
return GeneratedCode(_ctx)
[docs]
@typecheck
async def id(self) -> ModuleID:
"""The ID of the module
Note
----
This is lazyly evaluated, no operation is actually run.
Returns
-------
ModuleID
A reference to a 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)
@classmethod
def _id_type(cls) -> type[Scalar]:
return ModuleID
@classmethod
def _from_id_query_field(cls):
return "module"
[docs]
@typecheck
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.
"""
if hasattr(self, "_name"):
return self._name
_args: list[Arg] = []
_ctx = self._select("name", _args)
return await _ctx.execute(str)
[docs]
@typecheck
async def objects(self) -> list["TypeDef"]:
"""Objects served by this module"""
_args: list[Arg] = []
_ctx = self._select("objects", _args)
_ctx = TypeDef(_ctx)._select_multiple(
_kind="kind",
_optional="optional",
)
return await _ctx.execute(list[TypeDef])
[docs]
@typecheck
async def sdk(self) -> str:
"""The SDK used by this 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.
"""
if hasattr(self, "_sdk"):
return self._sdk
_args: list[Arg] = []
_ctx = self._select("sdk", _args)
return await _ctx.execute(str)
[docs]
@typecheck
async def sdk_runtime(self) -> str:
"""The SDK runtime module image 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.
"""
if hasattr(self, "_sdk_runtime"):
return self._sdk_runtime
_args: list[Arg] = []
_ctx = self._select("sdkRuntime", _args)
return await _ctx.execute(str)
[docs]
@typecheck
async def serve(
self,
*,
environment: Optional[Sequence[ModuleEnvironmentVariable]] = None,
) -> Optional[Void]:
"""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
-------
Optional[Void]
The absense 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.
"""
if hasattr(self, "_serve"):
return self._serve
_args = [
Arg("environment", environment, None),
]
_ctx = self._select("serve", _args)
return await _ctx.execute(Optional[Void])
[docs]
@typecheck
def source_directory(self) -> Directory:
"""The directory containing the module's source code"""
_args: list[Arg] = []
_ctx = self._select("sourceDirectory", _args)
return Directory(_ctx)
[docs]
@typecheck
async def source_directory_sub_path(self) -> str:
"""The module's subpath within the source 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.
"""
if hasattr(self, "_source_directory_sub_path"):
return self._source_directory_sub_path
_args: list[Arg] = []
_ctx = self._select("sourceDirectorySubPath", _args)
return await _ctx.execute(str)
[docs]
@typecheck
def with_object(self, object: "TypeDef") -> "Module":
"""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_(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)
class ObjectTypeDef(Type):
"""A definition of a custom object defined in a Module."""
[docs]
@typecheck
async def description(self) -> Optional[str]:
"""The doc string for the object, if any
Returns
-------
Optional[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(Optional[str])
[docs]
@typecheck
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_multiple(
_description="description",
_name="name",
)
return await _ctx.execute(list[FieldTypeDef])
[docs]
@typecheck
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_multiple(
_call="call",
_description="description",
_name="name",
)
return await _ctx.execute(list[Function])
[docs]
@typecheck
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)
class Port(Type):
"""A port exposed by a container."""
__slots__ = (
"_description",
"_port",
"_protocol",
)
_description: Optional[str]
_port: Optional[int]
_protocol: Optional[NetworkProtocol]
[docs]
@typecheck
async def description(self) -> Optional[str]:
"""The port description.
Returns
-------
Optional[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.
"""
if hasattr(self, "_description"):
return self._description
_args: list[Arg] = []
_ctx = self._select("description", _args)
return await _ctx.execute(Optional[str])
[docs]
@typecheck
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.
"""
if hasattr(self, "_port"):
return self._port
_args: list[Arg] = []
_ctx = self._select("port", _args)
return await _ctx.execute(int)
[docs]
@typecheck
async def protocol(self) -> NetworkProtocol:
"""The transport layer network 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.
"""
if hasattr(self, "_protocol"):
return self._protocol
_args: list[Arg] = []
_ctx = self._select("protocol", _args)
return await _ctx.execute(NetworkProtocol)
class Client(Root):
[docs]
@typecheck
def cache_volume(self, key: str) -> CacheVolume:
"""Constructs a cache volume for a given cache key.
Parameters
----------
key:
A string identifier to target this cache volume (e.g., "modules-
cache").
"""
_args = [
Arg("key", key),
]
_ctx = self._select("cacheVolume", _args)
return CacheVolume(_ctx)
[docs]
@typecheck
async def check_version_compatibility(self, version: str) -> bool:
"""Checks if the current Dagger Engine is compatible with an SDK's
required version.
Parameters
----------
version:
The SDK's required version.
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 = [
Arg("version", version),
]
_ctx = self._select("checkVersionCompatibility", _args)
return await _ctx.execute(bool)
[docs]
@typecheck
def container(
self,
*,
id: Optional[ContainerID] = None,
platform: Optional[Platform] = None,
) -> Container:
"""Loads a container from ID.
Null ID returns an empty container (scratch).
Optional platform argument initializes new containers to execute and
publish as that platform.
Platform defaults to that of the builder's host.
"""
_args = [
Arg("id", id, None),
Arg("platform", platform, None),
]
_ctx = self._select("container", _args)
return Container(_ctx)
[docs]
@typecheck
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]
@typecheck
def current_module(self) -> Module:
"""The module currently being served in the session, if any."""
_args: list[Arg] = []
_ctx = self._select("currentModule", _args)
return Module(_ctx)
[docs]
@typecheck
def directory(
self,
*,
id: Optional[DirectoryID] = None,
) -> Directory:
"""Load a directory by ID. No argument produces an empty directory."""
_args = [
Arg("id", id, None),
]
_ctx = self._select("directory", _args)
return Directory(_ctx)
[docs]
@typecheck
def file(self, id: FileID) -> File:
"""Loads a file by ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("file", _args)
return File(_ctx)
[docs]
@typecheck
def function(self, id: FunctionID) -> Function:
"""Load a function by ID"""
_args = [
Arg("id", id),
]
_ctx = self._select("function", _args)
return Function(_ctx)
[docs]
@typecheck
def generated_code(
self,
*,
id: Optional[GeneratedCodeID] = None,
) -> GeneratedCode:
"""Load GeneratedCode by ID, or create a new one if id is unset."""
_args = [
Arg("id", id, None),
]
_ctx = self._select("generatedCode", _args)
return GeneratedCode(_ctx)
[docs]
@typecheck
def git(
self,
url: str,
*,
keep_git_dir: Optional[bool] = None,
experimental_service_host: Optional[Container] = 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:
Set to true to keep .git directory.
experimental_service_host:
A service which must be started before the repo is fetched.
"""
_args = [
Arg("url", url),
Arg("keepGitDir", keep_git_dir, None),
Arg("experimentalServiceHost", experimental_service_host, None),
]
_ctx = self._select("git", _args)
return GitRepository(_ctx)
[docs]
@typecheck
def host(self) -> Host:
"""Queries the host environment."""
_args: list[Arg] = []
_ctx = self._select("host", _args)
return Host(_ctx)
[docs]
@typecheck
def http(
self,
url: str,
*,
experimental_service_host: Optional[Container] = 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]
@typecheck
def module(self, *, id: Optional[ModuleID] = None) -> Module:
"""Load a module by ID, or create a new one if id is unset."""
_args = [
Arg("id", id, None),
]
_ctx = self._select("module", _args)
return Module(_ctx)
[docs]
@typecheck
def new_function(self, name: str, return_type: "TypeDef") -> Function:
"""Create a new function from the provided definition."""
_args = [
Arg("name", name),
Arg("returnType", return_type),
]
_ctx = self._select("newFunction", _args)
return Function(_ctx)
[docs]
@typecheck
def pipeline(
self,
name: str,
*,
description: Optional[str] = None,
labels: Optional[Sequence[PipelineLabel]] = None,
) -> "Client":
"""Creates a named sub-pipeline.
Parameters
----------
name:
Pipeline name.
description:
Pipeline description.
labels:
Pipeline labels.
"""
_args = [
Arg("name", name),
Arg("description", description, None),
Arg("labels", labels, None),
]
_ctx = self._select("pipeline", _args)
return Client(_ctx)
[docs]
@typecheck
def secret(self, id: SecretID) -> "Secret":
"""Loads a secret from its ID."""
_args = [
Arg("id", id),
]
_ctx = self._select("secret", _args)
return Secret(_ctx)
[docs]
@typecheck
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]
@typecheck
def socket(self, *, id: Optional[SocketID] = None) -> "Socket":
"""Loads a socket by its ID."""
_args = [
Arg("id", id, None),
]
_ctx = self._select("socket", _args)
return Socket(_ctx)
[docs]
@typecheck
def type_def(self, *, id: Optional[TypeDefID] = None) -> "TypeDef":
_args = [
Arg("id", id, None),
]
_ctx = self._select("typeDef", _args)
return TypeDef(_ctx)
[docs]
def with_(self, cb: Callable[["Client"], "Client"]) -> "Client":
"""Call the provided callable with current Client.
This is useful for reusability and readability by not breaking the calling chain.
"""
return cb(self)
class Secret(Type):
"""A reference to a secret value, which can be handled more safely
than the value itself."""
[docs]
@typecheck
async def id(self) -> SecretID:
"""The identifier for this secret.
Note
----
This is lazyly evaluated, no operation is actually run.
Returns
-------
SecretID
A unique identifier for a 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)
@classmethod
def _id_type(cls) -> type[Scalar]:
return SecretID
@classmethod
def _from_id_query_field(cls):
return "secret"
[docs]
@typecheck
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)
class Socket(Type):
[docs]
@typecheck
async def id(self) -> SocketID:
"""The content-addressed identifier of the socket.
Note
----
This is lazyly evaluated, no operation is actually run.
Returns
-------
SocketID
A content-addressed socket identifier.
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)
@classmethod
def _id_type(cls) -> type[Scalar]:
return SocketID
@classmethod
def _from_id_query_field(cls):
return "socket"
class TypeDef(Type):
"""A definition of a parameter or return type in a Module."""
__slots__ = (
"_kind",
"_optional",
)
_kind: Optional[TypeDefKind]
_optional: Optional[bool]
[docs]
@typecheck
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]
@typecheck
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]
@typecheck
async def id(self) -> TypeDefID:
"""Note
----
This is lazyly evaluated, no operation is actually run.
Returns
-------
TypeDefID
A reference to a 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)
@classmethod
def _id_type(cls) -> type[Scalar]:
return TypeDefID
@classmethod
def _from_id_query_field(cls):
return "typeDef"
[docs]
@typecheck
async def kind(self) -> Optional[TypeDefKind]:
"""The kind of type this is (e.g. primitive, list, object)
Returns
-------
Optional[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.
"""
if hasattr(self, "_kind"):
return self._kind
_args: list[Arg] = []
_ctx = self._select("kind", _args)
return await _ctx.execute(Optional[TypeDefKind])
[docs]
@typecheck
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.
"""
if hasattr(self, "_optional"):
return self._optional
_args: list[Arg] = []
_ctx = self._select("optional", _args)
return await _ctx.execute(bool)
[docs]
@typecheck
def with_field(
self,
name: str,
type_def: "TypeDef",
*,
description: Optional[str] = None,
) -> "TypeDef":
"""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
"""
_args = [
Arg("name", name),
Arg("typeDef", type_def),
Arg("description", description, None),
]
_ctx = self._select("withField", _args)
return TypeDef(_ctx)
[docs]
@typecheck
def with_function(self, function: Function) -> "TypeDef":
"""Adds a function for an Object TypeDef, failing if the type is not an
object.
"""
_args = [
Arg("function", function),
]
_ctx = self._select("withFunction", _args)
return TypeDef(_ctx)
[docs]
@typecheck
def with_kind(self, kind: TypeDefKind) -> "TypeDef":
"""Sets the kind of the type."""
_args = [
Arg("kind", kind),
]
_ctx = self._select("withKind", _args)
return TypeDef(_ctx)
[docs]
@typecheck
def with_list_of(self, element_type: "TypeDef") -> "TypeDef":
"""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]
@typecheck
def with_object(
self,
name: str,
*,
description: Optional[str] = None,
) -> "TypeDef":
"""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, None),
]
_ctx = self._select("withObject", _args)
return TypeDef(_ctx)
[docs]
@typecheck
def with_optional(self, optional: bool) -> "TypeDef":
"""Sets whether this type can be set to null."""
_args = [
Arg("optional", optional),
]
_ctx = self._select("withOptional", _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)
_client = Client()
cache_volume = _client.cache_volume
check_version_compatibility = _client.check_version_compatibility
container = _client.container
current_function_call = _client.current_function_call
current_module = _client.current_module
default_platform = _client.default_platform
directory = _client.directory
file = _client.file
function = _client.function
generated_code = _client.generated_code
git = _client.git
host = _client.host
http = _client.http
module = _client.module
new_function = _client.new_function
pipeline = _client.pipeline
secret = _client.secret
set_secret = _client.set_secret
socket = _client.socket
type_def = _client.type_def
def default_client() -> Client:
"""Return the default client instance."""
return _client
__all__ = [
"BuildArg",
"CacheID",
"CacheSharingMode",
"CacheVolume",
"Client",
"Container",
"ContainerID",
"Directory",
"DirectoryID",
"EnvVariable",
"FieldTypeDef",
"File",
"FileID",
"Function",
"FunctionArg",
"FunctionCall",
"FunctionCallArgValue",
"FunctionCallInput",
"FunctionID",
"GeneratedCode",
"GeneratedCodeID",
"GitRef",
"GitRepository",
"Host",
"ImageLayerCompression",
"ImageMediaTypes",
"JSON",
"Label",
"ListTypeDef",
"Module",
"ModuleEnvironmentVariable",
"ModuleID",
"NetworkProtocol",
"ObjectTypeDef",
"PipelineLabel",
"Platform",
"Port",
"Secret",
"SecretID",
"Socket",
"SocketID",
"TypeDef",
"TypeDefID",
"TypeDefKind",
"Void",
"cache_volume",
"check_version_compatibility",
"container",
"current_function_call",
"current_module",
"default_client",
"default_platform",
"directory",
"file",
"function",
"generated_code",
"git",
"host",
"http",
"module",
"new_function",
"pipeline",
"secret",
"set_secret",
"socket",
"type_def",
]