coxbuild package

Subpackages

Submodules

coxbuild.configuration module

class coxbuild.configuration.Configuration(name: str = '', data: Optional[dict[str, Any]] = None)

Bases: object

Build configuration, entry key is case-insensitive.

copy() → coxbuild.configuration.Configuration

Copy configuration.

env()

Get environ section.

envbuild()

Get environ:coxbuild section.

get(key: str) → Optional[Any]
loadenv()

Load environ section from os.

section(name: str)

Get sub-section.

name: sub-section name

class coxbuild.configuration.ConfigurationAccessor(config: coxbuild.configuration.Configuration)

Bases: object

Configuration accessor.

get(key: str) → Any

Get configuration value (or fallback).

getPath(key: str) → pathlib.Path | None

Get configuration value (or fallback) as Path.

coxbuild.exceptions module

exception coxbuild.exceptions.CoxbuildException(message: str, cause: Optional[Exception] = None)

Bases: Exception

Common base exception for all coxbuild exceptions.

exception coxbuild.exceptions.CoxbuildRuntimeException(message: str, cause: Optional[Exception] = None)

Bases: coxbuild.exceptions.CoxbuildException

Base exception for runtime errors.

exception coxbuild.exceptions.CoxbuildSchemaException(message: str, cause: Optional[Exception] = None)

Bases: coxbuild.exceptions.CoxbuildException

Base exception for schema errors.

coxbuild.hooks module

class coxbuild.hooks.Hook(hook: Callable)

Bases: object

Hook.

hook: Callable

coxbuild.invocation module

class coxbuild.invocation.CommandExecutionArgs(cmds: list[str], env: Optional[dict[str, str]] = None, cwd: Optional[pathlib.Path] = None, timeout: Optional[datetime.timedelta] = None, input: Optional[str] = None, shell: bool = False, pipe: bool = False)

Bases: object

Arguments for command execution.

cmds: list[str]

command and arguments

cwd: pathlib.Path | None = None

current working directory

env: dict[str, str] | None = None

environ

input: str | None = None

text for stdin

pipe: bool = False

pipe and collect stdout and stderr

run(retry: int = 0, fail: bool = False) → coxbuild.invocation.CommandExecutionResult

Run command.

retry: the number of times to retry when failing fail: do not raise exception when the final result fails

shell: bool = False

use system shell

timeout: datetime.timedelta | None = None

maximum execution duration

exception coxbuild.invocation.CommandExecutionException(result: coxbuild.invocation.CommandExecutionResult, error: str = '')

Bases: coxbuild.exceptions.CoxbuildRuntimeException

Exception for command execution.

class coxbuild.invocation.CommandExecutionResult(args: coxbuild.invocation.CommandExecutionArgs, duration: datetime.timedelta = <factory>, code: typing.Optional[int] = None, stdout: str = '', stderr: str = '')

Bases: object

Result for command execution.

args: coxbuild.invocation.CommandExecutionArgs

arguments for command execution

code: int | None = None

exit code, None for timeout

property description

Return result’s description string.

duration: datetime.timedelta

execution duration

ensure()
stderr: str = ''

stderr in text (if pipe)

stdout: str = ''

stdout in text (if pipe)

property timeout: bool

Return if execution timeout.

coxbuild.invocation.execmd(args: coxbuild.invocation.CommandExecutionArgs)

Execute command and get result.

coxbuild.invocation.run(cmds: list[str], env: Optional[dict[str, str]] = None, cwd: Optional[pathlib.Path] = None, timeout: Optional[datetime.timedelta] = None, input: Optional[str] = None, shell: bool = False, pipe: bool = False, retry: int = 0, fail: bool = False) → coxbuild.invocation.CommandExecutionResult

Run command.

cmds: command and argument env: environ cwd: current working directory timeout: maximum execution duration input: text for stdin shell: use system shell pipe: pipe and collect stdout and stderr retry: the number of times to retry when failing fail: do not raise exception when the final result fails

coxbuild.managers module

class coxbuild.managers.Manager(extensions: dict[str, coxbuild.extensions.Extension] = <factory>, configBuilders: coxbuild.configurations.builders.ConfigurationBuilderCollection = <factory>)

Bases: object

Manage extensions and execution.

configBuilders: coxbuild.configurations.builders.ConfigurationBuilderCollection
copy() → coxbuild.managers.Manager

Copy manager.

execute(*tasks: str) → coxbuild.pipelines.PipelineResult

Execute tasks.

async executeAsync(*tasks: str) → coxbuild.pipelines.PipelineResult

Execute tasks asynchronously.

extensions: dict[str, coxbuild.extensions.Extension]
register(ext: coxbuild.extensions.Extension)

Register extension.

unregister(ext: coxbuild.extensions.Extension | str)

Unregister extension.

coxbuild.pipelines module

class coxbuild.pipelines.Pipeline(tasks: dict[str, coxbuild.tasks.Task] = <factory>, hooks: list[coxbuild.pipelines.PipelineHook] = <factory>)

Bases: object

Pipeline to run tasks.

after(body: Union[Callable[[coxbuild.pipelines.PipelineContext, coxbuild.pipelines.PipelineResult], Optional[Awaitable]], coxbuild.pipelines.PipelineAfterHook]) → coxbuild.pipelines.PipelineAfterHook

Add after hook.

afterTask(body: Union[Callable[[coxbuild.tasks.TaskContext, coxbuild.tasks.TaskResult], Optional[Union[Awaitable[bool], bool]]], coxbuild.pipelines.PipelineAfterTaskHook]) → coxbuild.pipelines.PipelineAfterTaskHook

Add after task hook.

before(body: Union[Callable[[coxbuild.pipelines.PipelineContext], Optional[Union[Awaitable[bool], bool]]], coxbuild.pipelines.PipelineBeforeHook]) → coxbuild.pipelines.PipelineBeforeHook

Add before hook.

beforeTask(body: Union[Callable[[coxbuild.tasks.TaskContext], Optional[Union[Awaitable[bool], bool]]], coxbuild.pipelines.PipelineBeforeTaskHook]) → coxbuild.pipelines.PipelineBeforeTaskHook

Add before task hook.

copy() → coxbuild.pipelines.Pipeline

Copy pipeline.

hook(hook: coxbuild.pipelines.PipelineHook) → None

Hook pipeline.

thook: the hook

hooks: list[coxbuild.pipelines.PipelineHook]

hook for pipeline

register(task: coxbuild.tasks.Task) → None

Register a task into pipeline.

task: task to register

tasks: dict[str, coxbuild.tasks.Task]

tasks in the pipeline

class coxbuild.pipelines.PipelineAfterHook(hook: Optional[Callable[[coxbuild.pipelines.PipelineContext, coxbuild.pipelines.PipelineResult], Optional[Awaitable]]] = None, extension: Extension | None = None)

Bases: coxbuild.pipelines.PipelineHook

After pipeline finished

hook: Optional[Callable[[coxbuild.pipelines.PipelineContext, coxbuild.pipelines.PipelineResult], Optional[Awaitable]]] = None
class coxbuild.pipelines.PipelineAfterTaskHook(hook: Optional[Callable[[coxbuild.tasks.TaskContext, coxbuild.tasks.TaskResult], Optional[Awaitable]]] = None, extension: Extension | None = None)

Bases: coxbuild.pipelines.PipelineHook

After task finished

hook: Optional[Callable[[coxbuild.tasks.TaskContext, coxbuild.tasks.TaskResult], Optional[Awaitable]]] = None
class coxbuild.pipelines.PipelineBeforeHook(hook: Optional[Callable[[coxbuild.pipelines.PipelineContext], Optional[Union[Awaitable[bool], bool]]]] = None, extension: Extension | None = None)

Bases: coxbuild.pipelines.PipelineHook

Before pipeline starting

hook: Optional[Callable[[coxbuild.pipelines.PipelineContext], Optional[Union[Awaitable[bool], bool]]]] = None
class coxbuild.pipelines.PipelineBeforeTaskHook(hook: Optional[Callable[[coxbuild.tasks.TaskContext], Optional[Union[Awaitable[bool], bool]]]] = None, extension: Extension | None = None)

Bases: coxbuild.pipelines.PipelineHook

Before task starting

hook: Optional[Callable[[coxbuild.tasks.TaskContext], Optional[Union[Awaitable[bool], bool]]]] = None
class coxbuild.pipelines.PipelineContext(pipeline: coxbuild.pipelines.Pipeline, tasks: list[coxbuild.tasks.TaskResult], unmatchedNames: list[str] = <factory>, config: typing.Optional[coxbuild.configurations.Configuration] = None)

Bases: object

Execution context for pipeline.

config: coxbuild.configurations.Configuration | None = None

configuration

pipeline: coxbuild.pipelines.Pipeline

pipeline

tasks: list[coxbuild.tasks.TaskResult]

tasks in the pipeline

unmatchedNames: list[str]

not found task names

class coxbuild.pipelines.PipelineHook(hook: Callable, extension: Extension | None = None)

Bases: coxbuild.hooks.Hook

Hooks for a pipeline.

extension: Extension | None = None
class coxbuild.pipelines.PipelineResult(duration: datetime.timedelta, tasks: list[coxbuild.tasks.TaskResult], exception: Optional[coxbuild.pipelines.PipelineRuntimeException] = None)

Bases: object

Execution result for pipeline.

property description: str

Return result’s description string.

duration: datetime.timedelta

execution duration

exception: coxbuild.pipelines.PipelineRuntimeException | None = None

exception when running

tasks: list[coxbuild.tasks.TaskResult]

tasks in the pipeline

class coxbuild.pipelines.PipelineRunner(pipeline: coxbuild.pipelines.Pipeline, tasks: list[coxbuild.tasks.Task], hooks: list[coxbuild.pipelines.PipelineHook], unmatchedNames: list[str])

Bases: coxbuild.runners.Runner

Runner for pipeline.

exception coxbuild.pipelines.PipelineRuntimeException(message: str, cause: Optional[Exception] = None)

Bases: coxbuild.exceptions.CoxbuildRuntimeException

Exception for pipeline running.

coxbuild.pipelines.afterPipeline(hook: Union[Callable[[coxbuild.pipelines.PipelineContext, coxbuild.pipelines.PipelineResult], Optional[Awaitable]], coxbuild.pipelines.PipelineAfterHook]) → coxbuild.pipelines.PipelineAfterHook

Decorator to configure after hook of a pipeline. Run after pipeline.

hook: hook function

coxbuild.pipelines.afterTask(hook: Union[Callable[[coxbuild.tasks.TaskContext, coxbuild.tasks.TaskResult], Optional[Awaitable]], coxbuild.pipelines.PipelineAfterTaskHook]) → coxbuild.pipelines.PipelineAfterTaskHook

Decorator to configure after task hook of a task. Run after task.

hook: hook function

coxbuild.pipelines.beforePipeline(hook: Union[Callable[[coxbuild.pipelines.PipelineContext], Optional[Union[Awaitable[bool], bool]]], coxbuild.pipelines.PipelineBeforeHook]) → coxbuild.pipelines.PipelineBeforeHook

Decorator to configure before hook of a pipeline. Run before pipeline.

hook: hook function

coxbuild.pipelines.beforeTask(hook: Union[Callable[[coxbuild.tasks.TaskContext], Optional[Union[Awaitable[bool], bool]]], coxbuild.pipelines.PipelineBeforeTaskHook]) → coxbuild.pipelines.PipelineBeforeTaskHook

Decorator to configure before task hook of a task. Run before task.

hook: hook function

coxbuild.runners module

class coxbuild.runners.Runner(func: Callable[[], Optional[Awaitable]])

Bases: object

Generic async runner.

coxbuild.runtime module

class coxbuild.runtime.ExecutionState(config: coxbuild.configurations.Configuration)

Bases: coxbuild.configurations.ConfigurationAccessor

property configuration: coxbuild.configurations.Configuration | None

Get configuration.

property event: EventContext | None

Get event context.

property handler: EventHandler | None

Get event handler.

property manager: Manager | None

Get manager.

property matchedTasks: list[Task]

Matched task names.

property pipeline: Pipeline | None

Get pipeline.

property service: Service | None

Get service.

property task: Task | None

Get task.

property unmatchedTasks: list[str]

Unmatched task names.

coxbuild.runtime.withConfig(task: Task) → Task

Decorator to add config argument to task context.

coxbuild.runtime.withEvent(task: Task) → Task

Decorator to add event argument to task context.

coxbuild.runtime.withExecutionState(task: Task) → Task

Decorator to add executionState argument to task context.

coxbuild.runtime.withHandler(task: Task) → Task

Decorator to add handler argument to task context.

coxbuild.runtime.withManager(task: Task) → Task

Decorator to add manager argument to task context.

coxbuild.runtime.withPipeline(task: Task) → Task

Decorator to add pipeline argument to task context.

coxbuild.runtime.withService(task: Task) → Task

Decorator to add service argument to task context.

coxbuild.runtime.withTask(task: Task) → Task

Decorator to add task argument to task context.

coxbuild.schema module

coxbuild.schema.ext(extension: module | str) → coxbuild.extensions.Extension

coxbuild.services module

class coxbuild.services.EventContext(args: list[typing.Any] = <factory>, kwds: dict[str, typing.Any] = <factory>)

Bases: object

Event occur information.

args: list[Any]

arguments

classmethod build(*args: Any, **kwds: Any) → coxbuild.services.EventContext
kwds: dict[str, Any]

keyword arguments

exception coxbuild.services.EventFailed(message: str, cause: Optional[Exception] = None)

Bases: coxbuild.exceptions.CoxbuildRuntimeException

Event failed.

class coxbuild.services.EventHandler(event: AsyncIterator[coxbuild.services.EventContext | None], handler: coxbuild.tasks.Task, safe: bool = False, name: str = '', extension: Extension | None = None)

Bases: object

Handler for a event.

event: AsyncIterator[coxbuild.services.EventContext | None]

event generator, when the event occurs, the awaitable return

extension: Extension | None = None
async handle(config: Optional[coxbuild.configurations.Configuration] = None)
handler: coxbuild.tasks.Task

event handler

name: str = ''

handler name

safe: bool = False

prevent exception

exception coxbuild.services.EventHandlerFailed(message: str, cause: Optional[Exception] = None)

Bases: coxbuild.exceptions.CoxbuildRuntimeException

Event handler failed.

class coxbuild.services.Service(handlers: list[coxbuild.services.EventHandler] = <factory>)

Bases: object

Event-based service.

copy() → coxbuild.services.Service

Copy the service.

handlers: list[coxbuild.services.EventHandler]

handlers in the service

register(handler: coxbuild.services.EventHandler)

Register handler.

class coxbuild.services.ServiceContext(service: coxbuild.services.Service, config: Optional[coxbuild.configurations.Configuration] = None)

Bases: object

Service context.

config: coxbuild.configurations.Configuration | None = None

configuration

service: coxbuild.services.Service

service

class coxbuild.services.ServiceRunner(service: coxbuild.services.Service)

Bases: coxbuild.runners.Runner

coxbuild.services.on(event: Callable[[], Awaitable], safe: bool = False, name: Optional[str] = None)

Decorator to register event handler.

event: event generator, when the event occurs, the awaitable return handler: event handler repeat: repeat times, 0 for no-repeat, positive integer for finite repeat, negative integer for infinite repeat safe: prevent exception name: handler name, None to use function name

coxbuild.tasks module

class coxbuild.tasks.Task(name: str = 'default', body: typing.Optional[typing.Callable[[...], typing.Optional[typing.Awaitable]]] = None, description: str = '', deps: list['Task'] = <factory>, hooks: list[coxbuild.tasks.TaskHook] = <factory>, continueOnError: bool = False, extension: Extension | None = None)

Bases: object

Task.

after(body: Union[Callable[[coxbuild.tasks.TaskContext, coxbuild.tasks.TaskResult], Optional[Awaitable]], coxbuild.tasks.TaskAfterHook]) → coxbuild.tasks.TaskAfterHook

Add after hook.

before(body: Union[Callable[[coxbuild.tasks.TaskContext], Optional[Union[Awaitable[bool], bool]]], coxbuild.tasks.TaskBeforeHook]) → coxbuild.tasks.TaskBeforeHook

Add before hook.

body: Optional[Callable[[...], Optional[Awaitable]]] = None

task body

continueOnError: bool = False

continue execution on error

copy() → coxbuild.tasks.Task

Clone task.

depend(task: coxbuild.tasks.Task) → coxbuild.tasks.Task

Add dependency task.

deps: list['Task']

dependency task names

description: str = ''

task document

extension: Extension | None = None
hook(hook: coxbuild.tasks.TaskHook) → coxbuild.tasks.TaskHook

Add hook.

hooks: list[coxbuild.tasks.TaskHook]

task hooks

name: str = 'default'

task name

postcond(body: Union[Callable[[...], Union[Awaitable[bool], bool]], coxbuild.tasks.TaskPostconditionHook]) → coxbuild.tasks.TaskPostconditionHook

Add postcondition hook.

precond(body: Union[Callable[[...], Union[Awaitable[bool], bool]], coxbuild.tasks.TaskPreconditionHook]) → coxbuild.tasks.TaskPreconditionHook

Add precondition hook.

setup(body: Union[Callable[[...], Optional[Awaitable]], coxbuild.tasks.TaskSetupHook]) → coxbuild.tasks.TaskSetupHook

Add setup hook.

teardown(body: Union[Callable[[...], Optional[Awaitable]], coxbuild.tasks.TaskTeardownHook]) → coxbuild.tasks.TaskTeardownHook

Add teardown hook.

class coxbuild.tasks.TaskAfterHook(hook: Optional[Callable[[coxbuild.tasks.TaskContext, coxbuild.tasks.TaskResult], Optional[Awaitable]]] = None)

Bases: coxbuild.tasks.TaskHook

After task finished

hook: Optional[Callable[[coxbuild.tasks.TaskContext, coxbuild.tasks.TaskResult], Optional[Awaitable]]] = None
class coxbuild.tasks.TaskBeforeHook(hook: Optional[Callable[[coxbuild.tasks.TaskContext], Optional[Union[Awaitable[bool], bool]]]] = None)

Bases: coxbuild.tasks.TaskHook

Before task starting

hook: Optional[Callable[[coxbuild.tasks.TaskContext], Optional[Union[Awaitable[bool], bool]]]] = None
class coxbuild.tasks.TaskContext(task: coxbuild.tasks.Task, args: list[typing.Any] = <factory>, kwds: dict[str, typing.Any] = <factory>, config: typing.Optional[coxbuild.configurations.Configuration] = None)

Bases: object

Execution context for task.

args: list[Any]

arguments

config: coxbuild.configurations.Configuration | None = None

configuration

kwds: dict[str, Any]

keyword arguments

task: coxbuild.tasks.Task

running task

class coxbuild.tasks.TaskHook(hook: Callable)

Bases: coxbuild.hooks.Hook

Hooks for a task.

exception coxbuild.tasks.TaskPostConditionFailed(task: coxbuild.tasks.Task)

Bases: coxbuild.tasks.TaskRuntimeException

Exception to indicate postcondition of a task is failed.

class coxbuild.tasks.TaskPostconditionHook(hook: Optional[Callable[[...], Union[Awaitable[bool], bool]]] = None)

Bases: coxbuild.tasks.TaskHook

Condition to decide whether to run the task

hook: Optional[Callable[[...], Union[Awaitable[bool], bool]]] = None
class coxbuild.tasks.TaskPreconditionHook(hook: Optional[Callable[[...], Union[Awaitable[bool], bool]]] = None)

Bases: coxbuild.tasks.TaskHook

Condition to decide whether to run the task

hook: Optional[Callable[[...], Union[Awaitable[bool], bool]]] = None
class coxbuild.tasks.TaskResult(task: coxbuild.tasks.Task, duration: datetime.timedelta, exception: coxbuild.tasks.TaskRuntimeException | None)

Bases: object

Execution result for task.

property description: str

Return result’s description string.

duration: datetime.timedelta

execution duration

ensure() → None

Ensure the result is success, otherwise re-raise exception.

exception: coxbuild.tasks.TaskRuntimeException | None

exception when running

task: coxbuild.tasks.Task

task

class coxbuild.tasks.TaskRunner(task: coxbuild.tasks.Task, args: list[Any], kwds: dict[str, Any])

Bases: coxbuild.runners.Runner

Runner for task.

exception coxbuild.tasks.TaskRuntimeException(task: coxbuild.tasks.Task, error: str = '', cause: Optional[Exception] = None)

Bases: coxbuild.exceptions.CoxbuildRuntimeException

Exception to indicate task is running.

class coxbuild.tasks.TaskSetupHook(hook: Optional[Callable[[...], Optional[Awaitable]]] = None)

Bases: coxbuild.tasks.TaskHook

Setup before task running

hook: Optional[Callable[[...], Optional[Awaitable]]] = None
class coxbuild.tasks.TaskTeardownHook(hook: Optional[Callable[[...], Optional[Awaitable]]] = None)

Bases: coxbuild.tasks.TaskHook

Teardown after task running

hook: Optional[Callable[[...], Optional[Awaitable]]] = None
coxbuild.tasks.after(hook: Union[Callable[[coxbuild.tasks.TaskContext, coxbuild.tasks.TaskResult], Optional[Awaitable]], coxbuild.tasks.TaskAfterHook])

Decorator to configure after hook of the task.

coxbuild.tasks.asafter(hook: Union[Callable[[coxbuild.tasks.TaskContext, coxbuild.tasks.TaskResult], Optional[Awaitable]], coxbuild.tasks.TaskAfterHook])

Decorator to define after hook of a task. Run after task body.

hook: hook function

coxbuild.tasks.asbefore(hook: Union[Callable[[coxbuild.tasks.TaskContext], Optional[Union[Awaitable[bool], bool]]], coxbuild.tasks.TaskBeforeHook])

Decorator to define before hook of a task. Run before task body.

hook: hook function

coxbuild.tasks.aspostcond(predicate: Union[Callable[[...], bool], coxbuild.tasks.TaskPostconditionHook])

Decorator to define postcondition of the task. Check the task works well.

predicate: condition tester

coxbuild.tasks.asprecond(predicate: Union[Callable[[...], bool], coxbuild.tasks.TaskPreconditionHook])

Decorator to define precondition of a task. Decide whether to run a task.

predicate: condition tester

coxbuild.tasks.assetup(hook: Union[Callable[[...], Optional[Awaitable]], coxbuild.tasks.TaskSetupHook])

Decorator to define setup hook of a task. Run before task body.

hook: hook function

coxbuild.tasks.asteardown(hook: Union[Callable[[...], Optional[Awaitable]], coxbuild.tasks.TaskTeardownHook])

Decorator to define teardown hook of a task. Run after task body.

hook: hook function

coxbuild.tasks.before(hook: Union[Callable[[coxbuild.tasks.TaskContext], Union[Awaitable[bool], bool]], coxbuild.tasks.TaskBeforeHook])

Decorator to configure before hook of the task.

coxbuild.tasks.continueOnError(inner: coxbuild.tasks.Task) → coxbuild.tasks.Task

Decorator to continue task execution on error.

coxbuild.tasks.depend(*tasks: coxbuild.tasks.Task)

Decorator to define dependencies of a task.

names: task names or task instances

coxbuild.tasks.group(*names: str)

Decorator to set task group.

coxbuild.tasks.named(name: str)

Decorator to set task name.

coxbuild.tasks.postcond(hook: Union[Callable[[...], Union[Awaitable[bool], bool]], coxbuild.tasks.TaskPostconditionHook])

Decorator to configure postcondition hook of the task.

coxbuild.tasks.precond(hook: Union[Callable[[...], Union[Awaitable[bool], bool]], coxbuild.tasks.TaskPreconditionHook])

Decorator to configure precondition hook of the task.

coxbuild.tasks.setup(hook: Union[Callable[[...], Optional[Awaitable]], coxbuild.tasks.TaskSetupHook])

Decorator to configure setup hook of the task.

coxbuild.tasks.task(body: Callable[[...], None]) → coxbuild.tasks.Task

Decorator to define a task.

coxbuild.tasks.taskconfigs(*args: Callable[[coxbuild.tasks.Task], coxbuild.tasks.Task]) → Callable[[coxbuild.tasks.Task], coxbuild.tasks.Task]

Combine multiple task configs.

taskconfig(a, b, c) is equivalent to c(b(a(task)))

coxbuild.tasks.teardown(hook: Union[Callable[[...], Optional[Awaitable]], coxbuild.tasks.TaskTeardownHook])

Decorator to configure teardown hook of the task.

Module contents

coxbuild.get_app_directory() → pathlib.Path

Get coxbuild root directory.

coxbuild.get_working_directory() → pathlib.Path

Get coxbuild working directory.