MyCode Project

Overview

  • Minimal starter template you can copy to bootstrap a new AGILab application.

  • Demonstrates the project layout expected by the platform (manager package, worker package, app_args definitions, Explore configuration) with minimal business logic so you can focus on custom code.

  • Ships with blank Streamlit forms and prompt files to illustrate where to plug in UI customisation and Experiment prompts.

Scientific placeholders

The template is intentionally lightweight, but many AGILab workflows ultimately fit the pattern of learning or calibrating a function \(f_\\theta\):

\[\begin{split}\\theta^* = \\arg\\min_{\\theta} \\; \\frac{1}{N} \\sum_{i=1}^{N} \\ell\\left(f_{\\theta}(x_i), y_i\\right)\end{split}\]

where \(\\ell\) is a task-dependent loss (regression, classification, imitation learning, etc.). You can use the MyCode skeleton to prototype the data loading, feature extraction, and artifact export around this core loop.

Manager (mycode.mycode)

  • Lightweight subclass of BaseWorker that shows how to wire argument handling, logging and dataframe export without heavy dependencies.

  • Provides the same from_toml / to_toml helpers as production projects so you can reuse the Execute page snippets verbatim.

Args (mycode.app_args)

  • Simple Pydantic model that mirrors the keys present in app_settings.toml.

  • Ideal starting point for capturing new configuration options; extend the model and the Streamlit form generated on the Execute page will pick them up.

Worker (mycode_worker.mycode_worker)

  • Skeleton worker that demonstrates the lifecycle hooks (start, work_init, run) required by the distributor.

  • Includes placeholder logic for dataset loading and result persistence—replace with your domain-specific processing steps.

Assets & Tests

  • app_test.py ensures the installer and worker skeleton keep working as the template evolves.

  • test/_test_* modules show how to unit-test managers and workers in isolation.

  • app_args_form.py provides the optional Streamlit form that mirrors the generated UI; tailor it when you need additional validation or widgets.

API Reference

mycode package diagram

Minimal manager implementation for the mycode sample project.

class mycode.mycode.Mycode(env, args=None, **kwargs)[source]

Bases: BaseWorker

Lightweight orchestration surface for the mycode example.

__init__(env, args=None, **kwargs)[source]
args_dump_mode: ClassVar[str] = 'json'
args_dumper: ClassVar[Callable[..., None] | None] = None
args_ensure_defaults: ClassVar[Callable[..., Any] | None] = None
args_loader: ClassVar[Callable[..., Any] | None] = None
args_merger: ClassVar[Callable[[Any, Optional[Any]], Any] | None] = None
as_dict()[source]
Return type:

dict[str, Any]

break()

Signal the service loop to exit on this worker.

Return type:

bool

static break_loop()

Signal the service loop to exit on this worker.

Return type:

bool

build_distribution(_workers=None)[source]
Return type:

Tuple[List[List], List[List[Tuple[int, int]]], str, str, str]

default_settings_path: ClassVar[str] = 'app_settings.toml'
default_settings_section: ClassVar[str] = 'args'
env: Optional[AgiEnv] = None
static expand(path, base_directory=None)

Expand a given path to an absolute path. :type path: str :param path: The path to expand. :type path: str :type base_directory: str, optional :param base_directory: The base directory to use for expanding the path. Defaults to None. :type base_directory: str, optional

Returns:

The expanded absolute path.

Return type:

str

Raises:

None –

Note

This method handles both Unix and Windows paths and expands ‘~’ notation to the user’s home directory.

static expand_and_join(path1, path2)

Join two paths after expanding the first path.

Parameters:

  • path1 (str) – The first path to expand and join.

  • path2 (str) – The second path to join with the expanded first path.

Returns:

The joined path.

Return type:

str

classmethod from_toml(env, settings_path='app_settings.toml', section='args', **overrides)[source]
Return type:

Mycode

static loop(*, poll_interval=None)

Run a long-lived service loop on this worker until signalled to stop.

The derived worker can implement a loop method accepting either zero arguments or a single stop_event argument. When the method signature accepts stop_event (keyword stop_event or should_stop), the worker implementation is responsible for honouring the event. Otherwise the base implementation repeatedly invokes the method and sleeps for the configured poll interval between calls. Returning False from the worker method requests termination of the loop.

Return type:

Dict[str, Any]

managed_pc_home_suffix: ClassVar[str] = 'MyApp'
managed_pc_path_fields: ClassVar[tuple[str, ...]] = ()
static normalize_dataset_path(data_path)

Normalise any dataset directory input so workers rely on consistent paths.

Return type:

str

static pool_init(vars)[source]
Return type:

None

prepare_output_dir(root, *, subdir='dataframe', attribute='data_out', clean=True)

Create (and optionally reset) a deterministic output directory.

Return type:

Path

setup_args(args, *, env=None, error=None, output_field=None, output_subdir='dataframe', output_attr='data_out', output_clean=True, output_parents_up=0)
Return type:

Any

setup_data_directories(*, source_path, target_path=None, target_subdir='dataframe', reset_target=False)

Prepare normalised input/output dataset paths without relying on worker args.

Returns a namespace with the resolved input path (input_path), the normalised string used by downstream readers (normalized_input), the output directory as a Path (output_path), and its normalised string representation (normalized_output). Optionally clears and recreates the output directory.

Return type:

SimpleNamespace

static start(worker_inst)

Invoke the concrete worker’s start hook once initialised.

stop()[source]

Returns:

Return type:

None

to_toml(settings_path='app_settings.toml', section='args', create_missing=True)[source]
Return type:

None

verbose = 1
work_done(_)[source]
Return type:

None

work_pool(_=None)[source]
Return type:

None

worker_vars: dict[str, Any] = {}
class mycode.mycode.MycodeApp(env, args=None, **kwargs)[source]

Bases: Mycode

Alias retaining the historical suffix for compatibility.

__init__(env, args=None, **kwargs)
args_dump_mode: ClassVar[str] = 'json'
args_dumper: ClassVar[Callable[..., None] | None] = None
args_ensure_defaults: ClassVar[Callable[..., Any] | None] = None
args_loader: ClassVar[Callable[..., Any] | None] = None
args_merger: ClassVar[Callable[[Any, Optional[Any]], Any] | None] = None
as_dict()
Return type:

dict[str, Any]

break()

Signal the service loop to exit on this worker.

Return type:

bool

static break_loop()

Signal the service loop to exit on this worker.

Return type:

bool

build_distribution(_workers=None)
Return type:

Tuple[List[List], List[List[Tuple[int, int]]], str, str, str]

default_settings_path: ClassVar[str] = 'app_settings.toml'
default_settings_section: ClassVar[str] = 'args'
env: Optional[AgiEnv] = None
static expand(path, base_directory=None)

Expand a given path to an absolute path. :type path: str :param path: The path to expand. :type path: str :type base_directory: str, optional :param base_directory: The base directory to use for expanding the path. Defaults to None. :type base_directory: str, optional

Returns:

The expanded absolute path.

Return type:

str

Raises:

None –

Note

This method handles both Unix and Windows paths and expands ‘~’ notation to the user’s home directory.

static expand_and_join(path1, path2)

Join two paths after expanding the first path.

Parameters:

  • path1 (str) – The first path to expand and join.

  • path2 (str) – The second path to join with the expanded first path.

Returns:

The joined path.

Return type:

str

classmethod from_toml(env, settings_path='app_settings.toml', section='args', **overrides)
Return type:

Mycode

static loop(*, poll_interval=None)

Run a long-lived service loop on this worker until signalled to stop.

The derived worker can implement a loop method accepting either zero arguments or a single stop_event argument. When the method signature accepts stop_event (keyword stop_event or should_stop), the worker implementation is responsible for honouring the event. Otherwise the base implementation repeatedly invokes the method and sleeps for the configured poll interval between calls. Returning False from the worker method requests termination of the loop.

Return type:

Dict[str, Any]

managed_pc_home_suffix: ClassVar[str] = 'MyApp'
managed_pc_path_fields: ClassVar[tuple[str, ...]] = ()
static normalize_dataset_path(data_path)

Normalise any dataset directory input so workers rely on consistent paths.

Return type:

str

static pool_init(vars)
Return type:

None

prepare_output_dir(root, *, subdir='dataframe', attribute='data_out', clean=True)

Create (and optionally reset) a deterministic output directory.

Return type:

Path

setup_args(args, *, env=None, error=None, output_field=None, output_subdir='dataframe', output_attr='data_out', output_clean=True, output_parents_up=0)
Return type:

Any

setup_data_directories(*, source_path, target_path=None, target_subdir='dataframe', reset_target=False)

Prepare normalised input/output dataset paths without relying on worker args.

Returns a namespace with the resolved input path (input_path), the normalised string used by downstream readers (normalized_input), the output directory as a Path (output_path), and its normalised string representation (normalized_output). Optionally clears and recreates the output directory.

Return type:

SimpleNamespace

static start(worker_inst)

Invoke the concrete worker’s start hook once initialised.

stop()

Returns:

Return type:

None

to_toml(settings_path='app_settings.toml', section='args', create_missing=True)
Return type:

None

verbose = 1
work_done(_)
Return type:

None

work_pool(_=None)
Return type:

None

worker_vars: dict[str, Any] = {}
mycode class diagram

Argument management helpers for the mycode sample project.

mycode.app_args.ArgsModel

alias of MycodeArgs

mycode.app_args.ArgsOverrides

alias of MycodeArgsTD

class mycode.app_args.MycodeArgs(**data)[source]

Bases: BaseModel

Runtime parameters for the mycode application.

__init__(**data)

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

classmethod construct(_fields_set=None, **values)
Return type:

Self

copy(*, include=None, exclude=None, update=None, deep=False)

Returns a copy of the model.

!!! warning “Deprecated”

This method is now deprecated; use model_copy instead.

If you need include or exclude, use:

`python {test="skip" lint="skip"} data = self.model_dump(include=include, exclude=exclude, round_trip=True) data = {**data, **(update or {})} copied = self.model_validate(data) `

Parameters:

  • include (Set[int] | Set[str] | Mapping[int, Any] | Mapping[str, Any] | None) – Optional set or mapping specifying which fields to include in the copied model.

  • exclude (Set[int] | Set[str] | Mapping[int, Any] | Mapping[str, Any] | None) – Optional set or mapping specifying which fields to exclude in the copied model.

  • update (Optional[Dict[str, Any]]) – Optional dictionary of field-value pairs to override field values in the copied model.

  • deep (bool) – If True, the values of fields that are Pydantic models will be deep-copied.

Return type:

Self

Returns:

A copy of the model with included, excluded and updated fields as specified.

data_in: Path
data_out: Path
dict(*, include=None, exclude=None, by_alias=False, exclude_unset=False, exclude_defaults=False, exclude_none=False)
Return type:

Dict[str, Any]

files: str
classmethod from_orm(obj)
Return type:

Self

json(*, include=None, exclude=None, by_alias=False, exclude_unset=False, exclude_defaults=False, exclude_none=False, encoder=PydanticUndefined, models_as_dict=PydanticUndefined, **dumps_kwargs)
Return type:

str

model_computed_fields = {}
model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

classmethod model_construct(_fields_set=None, **values)

Creates a new instance of the Model class with validated data.

Creates a new model setting __dict__ and __pydantic_fields_set__ from trusted or pre-validated data. Default values are respected, but no other validation is performed.

!!! note

model_construct() generally respects the model_config.extra setting on the provided model. That is, if model_config.extra == ‘allow’, then all extra passed values are added to the model instance’s __dict__ and __pydantic_extra__ fields. If model_config.extra == ‘ignore’ (the default), then all extra passed values are ignored. Because no validation is performed with a call to model_construct(), having model_config.extra == ‘forbid’ does not result in an error if extra values are passed, but they will be ignored.

Parameters:

  • _fields_set (set[str] | None) – A set of field names that were originally explicitly set during instantiation. If provided, this is directly used for the [model_fields_set][pydantic.BaseModel.model_fields_set] attribute. Otherwise, the field names from the values argument will be used.

  • values (Any) – Trusted or pre-validated data dictionary.

Return type:

Self

Returns:

A new instance of the Model class with validated data.

model_copy(*, update=None, deep=False)
!!! abstract “Usage Documentation”

[model_copy](../concepts/models.md#model-copy)

Returns a copy of the model.

!!! note

The underlying instance’s [__dict__][object.__dict__] attribute is copied. This might have unexpected side effects if you store anything in it, on top of the model fields (e.g. the value of [cached properties][functools.cached_property]).

Parameters:

  • update (Mapping[str, Any] | None) – Values to change/add in the new model. Note: the data is not validated before creating the new model. You should trust this data.

  • deep (bool) – Set to True to make a deep copy of the model.

Return type:

Self

Returns:

New model instance.

model_dump(*, mode='python', include=None, exclude=None, context=None, by_alias=None, exclude_unset=False, exclude_defaults=False, exclude_none=False, exclude_computed_fields=False, round_trip=False, warnings=True, fallback=None, serialize_as_any=False)
!!! abstract “Usage Documentation”

[model_dump](../concepts/serialization.md#python-mode)

Generate a dictionary representation of the model, optionally specifying which fields to include or exclude.

Parameters:

  • mode (Union[Literal['json', 'python'], str]) – The mode in which to_python should run. If mode is ‘json’, the output will only contain JSON serializable types. If mode is ‘python’, the output may contain non-JSON-serializable Python objects.

  • include (Union[set[int], set[str], Mapping[int, Union[set[int], set[str], Mapping[int, Union[IncEx, bool]], Mapping[str, Union[IncEx, bool]], bool]], Mapping[str, Union[set[int], set[str], Mapping[int, Union[IncEx, bool]], Mapping[str, Union[IncEx, bool]], bool]], None]) – A set of fields to include in the output.

  • exclude (Union[set[int], set[str], Mapping[int, Union[set[int], set[str], Mapping[int, Union[IncEx, bool]], Mapping[str, Union[IncEx, bool]], bool]], Mapping[str, Union[set[int], set[str], Mapping[int, Union[IncEx, bool]], Mapping[str, Union[IncEx, bool]], bool]], None]) – A set of fields to exclude from the output.

  • context (Any | None) – Additional context to pass to the serializer.

  • by_alias (bool | None) – Whether to use the field’s alias in the dictionary key if defined.

  • exclude_unset (bool) – Whether to exclude fields that have not been explicitly set.

  • exclude_defaults (bool) – Whether to exclude fields that are set to their default value.

  • exclude_none (bool) – Whether to exclude fields that have a value of None.

  • exclude_computed_fields (bool) – Whether to exclude computed fields. While this can be useful for round-tripping, it is usually recommended to use the dedicated round_trip parameter instead.

  • round_trip (bool) – If True, dumped values should be valid as input for non-idempotent types such as Json[T].

  • warnings (Union[bool, Literal['none', 'warn', 'error']]) – How to handle serialization errors. False/”none” ignores them, True/”warn” logs errors, “error” raises a [PydanticSerializationError][pydantic_core.PydanticSerializationError].

  • fallback (Optional[Callable[[Any], Any]]) – A function to call when an unknown value is encountered. If not provided, a [PydanticSerializationError][pydantic_core.PydanticSerializationError] error is raised.

  • serialize_as_any (bool) – Whether to serialize fields with duck-typing serialization behavior.

Return type:

dict[str, Any]

Returns:

A dictionary representation of the model.

model_dump_json(*, indent=None, ensure_ascii=False, include=None, exclude=None, context=None, by_alias=None, exclude_unset=False, exclude_defaults=False, exclude_none=False, exclude_computed_fields=False, round_trip=False, warnings=True, fallback=None, serialize_as_any=False)
!!! abstract “Usage Documentation”

[model_dump_json](../concepts/serialization.md#json-mode)

Generates a JSON representation of the model using Pydantic’s to_json method.

Parameters:

  • indent (int | None) – Indentation to use in the JSON output. If None is passed, the output will be compact.

  • ensure_ascii (bool) – If True, the output is guaranteed to have all incoming non-ASCII characters escaped. If False (the default), these characters will be output as-is.

  • include (Union[set[int], set[str], Mapping[int, Union[set[int], set[str], Mapping[int, Union[IncEx, bool]], Mapping[str, Union[IncEx, bool]], bool]], Mapping[str, Union[set[int], set[str], Mapping[int, Union[IncEx, bool]], Mapping[str, Union[IncEx, bool]], bool]], None]) – Field(s) to include in the JSON output.

  • exclude (Union[set[int], set[str], Mapping[int, Union[set[int], set[str], Mapping[int, Union[IncEx, bool]], Mapping[str, Union[IncEx, bool]], bool]], Mapping[str, Union[set[int], set[str], Mapping[int, Union[IncEx, bool]], Mapping[str, Union[IncEx, bool]], bool]], None]) – Field(s) to exclude from the JSON output.

  • context (Any | None) – Additional context to pass to the serializer.

  • by_alias (bool | None) – Whether to serialize using field aliases.

  • exclude_unset (bool) – Whether to exclude fields that have not been explicitly set.

  • exclude_defaults (bool) – Whether to exclude fields that are set to their default value.

  • exclude_none (bool) – Whether to exclude fields that have a value of None.

  • exclude_computed_fields (bool) – Whether to exclude computed fields. While this can be useful for round-tripping, it is usually recommended to use the dedicated round_trip parameter instead.

  • round_trip (bool) – If True, dumped values should be valid as input for non-idempotent types such as Json[T].

  • warnings (Union[bool, Literal['none', 'warn', 'error']]) – How to handle serialization errors. False/”none” ignores them, True/”warn” logs errors, “error” raises a [PydanticSerializationError][pydantic_core.PydanticSerializationError].

  • fallback (Optional[Callable[[Any], Any]]) – A function to call when an unknown value is encountered. If not provided, a [PydanticSerializationError][pydantic_core.PydanticSerializationError] error is raised.

  • serialize_as_any (bool) – Whether to serialize fields with duck-typing serialization behavior.

Return type:

str

Returns:

A JSON string representation of the model.

property model_extra: dict[str, Any] | None

Get extra fields set during validation.

Returns:

A dictionary of extra fields, or None if config.extra is not set to “allow”.

model_fields = {'data_in': FieldInfo(annotation=Path, required=False, default_factory=<lambda>), 'data_out': FieldInfo(annotation=Path, required=False, default_factory=<lambda>), 'files': FieldInfo(annotation=str, required=False, default='*'), 'nfile': FieldInfo(annotation=int, required=False, default=1), 'nread': FieldInfo(annotation=int, required=False, default=0), 'nskip': FieldInfo(annotation=int, required=False, default=0), 'reset_target': FieldInfo(annotation=bool, required=False, default=False)}
property model_fields_set: set[str]

Returns the set of fields that have been explicitly set on this model instance.

Returns:

A set of strings representing the fields that have been set,

i.e. that were not filled from defaults.

classmethod model_json_schema(by_alias=True, ref_template='#/$defs/{model}', schema_generator=<class 'pydantic.json_schema.GenerateJsonSchema'>, mode='validation', *, union_format='any_of')

Generates a JSON schema for a model class.

Parameters:

  • by_alias (bool) – Whether to use attribute aliases or not.

  • ref_template (str) – The reference template.

  • union_format (Literal['any_of', 'primitive_type_array']) –

    The format to use when combining schemas from unions together. Can be one of:

    keyword to combine schemas (the default). - ‘primitive_type_array’: Use the [type](https://json-schema.org/understanding-json-schema/reference/type) keyword as an array of strings, containing each type of the combination. If any of the schemas is not a primitive type (string, boolean, null, integer or number) or contains constraints/metadata, falls back to any_of.

  • schema_generator (type[GenerateJsonSchema]) – To override the logic used to generate the JSON schema, as a subclass of GenerateJsonSchema with your desired modifications

  • mode (Literal['validation', 'serialization']) – The mode in which to generate the schema.

Return type:

dict[str, Any]

Returns:

The JSON schema for the given model class.

classmethod model_parametrized_name(params)

Compute the class name for parametrizations of generic classes.

This method can be overridden to achieve a custom naming scheme for generic BaseModels.

Parameters:

params (tuple[type[Any], ...]) – Tuple of types of the class. Given a generic class Model with 2 type variables and a concrete model Model[str, int], the value (str, int) would be passed to params.

Return type:

str

Returns:

String representing the new class where params are passed to cls as type variables.

Raises:

TypeError – Raised when trying to generate concrete names for non-generic models.

model_post_init(context, /)

Override this method to perform additional initialization after __init__ and model_construct. This is useful if you want to do some validation that requires the entire model to be initialized.

Return type:

None

classmethod model_rebuild(*, force=False, raise_errors=True, _parent_namespace_depth=2, _types_namespace=None)

Try to rebuild the pydantic-core schema for the model.

This may be necessary when one of the annotations is a ForwardRef which could not be resolved during the initial attempt to build the schema, and automatic rebuilding fails.

Parameters:

  • force (bool) – Whether to force the rebuilding of the model schema, defaults to False.

  • raise_errors (bool) – Whether to raise errors, defaults to True.

  • _parent_namespace_depth (int) – The depth level of the parent namespace, defaults to 2.

  • _types_namespace (Mapping[str, Any] | None) – The types namespace, defaults to None.

Return type:

bool | None

Returns:

Returns None if the schema is already “complete” and rebuilding was not required. If rebuilding _was_ required, returns True if rebuilding was successful, otherwise False.

classmethod model_validate(obj, *, strict=None, extra=None, from_attributes=None, context=None, by_alias=None, by_name=None)

Validate a pydantic model instance.

Parameters:

  • obj (Any) – The object to validate.

  • strict (bool | None) – Whether to enforce types strictly.

  • extra (Optional[Literal['allow', 'ignore', 'forbid']]) – Whether to ignore, allow, or forbid extra data during model validation. See the [extra configuration value][pydantic.ConfigDict.extra] for details.

  • from_attributes (bool | None) – Whether to extract data from object attributes.

  • context (Any | None) – Additional context to pass to the validator.

  • by_alias (bool | None) – Whether to use the field’s alias when validating against the provided input data.

  • by_name (bool | None) – Whether to use the field’s name when validating against the provided input data.

Raises:

ValidationError – If the object could not be validated.

Return type:

Self

Returns:

The validated model instance.

classmethod model_validate_json(json_data, *, strict=None, extra=None, context=None, by_alias=None, by_name=None)
!!! abstract “Usage Documentation”

[JSON Parsing](../concepts/json.md#json-parsing)

Validate the given JSON data against the Pydantic model.

Parameters:

  • json_data (str | bytes | bytearray) – The JSON data to validate.

  • strict (bool | None) – Whether to enforce types strictly.

  • extra (Optional[Literal['allow', 'ignore', 'forbid']]) – Whether to ignore, allow, or forbid extra data during model validation. See the [extra configuration value][pydantic.ConfigDict.extra] for details.

  • context (Any | None) – Extra variables to pass to the validator.

  • by_alias (bool | None) – Whether to use the field’s alias when validating against the provided input data.

  • by_name (bool | None) – Whether to use the field’s name when validating against the provided input data.

Return type:

Self

Returns:

The validated Pydantic model.

Raises:

ValidationError – If json_data is not a JSON string or the object could not be validated.

classmethod model_validate_strings(obj, *, strict=None, extra=None, context=None, by_alias=None, by_name=None)

Validate the given object with string data against the Pydantic model.

Parameters:

  • obj (Any) – The object containing string data to validate.

  • strict (bool | None) – Whether to enforce types strictly.

  • extra (Optional[Literal['allow', 'ignore', 'forbid']]) – Whether to ignore, allow, or forbid extra data during model validation. See the [extra configuration value][pydantic.ConfigDict.extra] for details.

  • context (Any | None) – Extra variables to pass to the validator.

  • by_alias (bool | None) – Whether to use the field’s alias when validating against the provided input data.

  • by_name (bool | None) – Whether to use the field’s name when validating against the provided input data.

Return type:

Self

Returns:

The validated Pydantic model.

nfile: int
nread: int
nskip: int
classmethod parse_file(path, *, content_type=None, encoding='utf8', proto=None, allow_pickle=False)
Return type:

Self

classmethod parse_obj(obj)
Return type:

Self

classmethod parse_raw(b, *, content_type=None, encoding='utf8', proto=None, allow_pickle=False)
Return type:

Self

reset_target: bool
classmethod schema(by_alias=True, ref_template='#/$defs/{model}')
Return type:

Dict[str, Any]

classmethod schema_json(*, by_alias=True, ref_template='#/$defs/{model}', **dumps_kwargs)
Return type:

str

classmethod update_forward_refs(**localns)
Return type:

None

classmethod validate(value)
Return type:

Self

class mycode.app_args.MycodeArgsTD[source]

Bases: TypedDict

__init__(*args, **kwargs)
clear()

Remove all items from the dict.

copy()

Return a shallow copy of the dict.

data_in: str
data_out: str
files: str
classmethod fromkeys(iterable, value=None, /)

Create a new dictionary with keys from iterable and values set to value.

get(key, default=None, /)

Return the value for key if key is in the dictionary, else default.

items()

Return a set-like object providing a view on the dict’s items.

keys()

Return a set-like object providing a view on the dict’s keys.

nfile: int
nread: int
nskip: int
pop(k[, d]) → v, remove specified key and return the corresponding value.

If the key is not found, return the default if given; otherwise, raise a KeyError.

popitem()

Remove and return a (key, value) pair as a 2-tuple.

Pairs are returned in LIFO (last-in, first-out) order. Raises KeyError if the dict is empty.

reset_target: bool
setdefault(key, default=None, /)

Insert key with a value of default if key is not in the dictionary.

Return the value for key if key is in the dictionary, else default.

update([E, ]**F) → None.  Update D from mapping/iterable E and F.

If E is present and has a .keys() method, then does: for k in E.keys(): D[k] = E[k] If E is present and lacks a .keys() method, then does: for k, v in E: D[k] = v In either case, this is followed by: for k in F: D[k] = F[k]

values()

Return an object providing a view on the dict’s values.

mycode.app_args.dump_args(args, settings_path, *, section='args', create_missing=True)[source]
Return type:

None

mycode.app_args.ensure_defaults(args, **_)[source]
Return type:

MycodeArgs

mycode.app_args.load_args(settings_path, *, section='args')[source]
Return type:

MycodeArgs

mycode.app_args.merge_args(base, overrides=None)[source]
Return type:

MycodeArgs

mycode args class diagram mycode worker package diagram

Default worker implementation for the mycode project.

This worker simply inherits the PolarsWorker so that projects with minimal requirements still provide a concrete worker class. Downstream installers rely on the class name MycodeWorker to determine the runtime bundle to ship.

class mycode_worker.mycode_worker.MycodeWorker[source]

Bases: PolarsWorker

Polars worker used by the mycode sample application.

args_dump_mode: ClassVar[str] = 'json'
args_dumper: ClassVar[Callable[..., None] | None] = None
args_ensure_defaults: ClassVar[Callable[..., Any] | None] = None
args_loader: ClassVar[Callable[..., Any] | None] = None
args_merger: ClassVar[Callable[[Any, Optional[Any]], Any] | None] = None
as_dict(mode=None)
Return type:

dict[str, Any]

break()

Signal the service loop to exit on this worker.

Return type:

bool

static break_loop()

Signal the service loop to exit on this worker.

Return type:

bool

default_settings_path: ClassVar[str] = 'app_settings.toml'
default_settings_section: ClassVar[str] = 'args'
env: Optional[AgiEnv] = None
static expand(path, base_directory=None)

Expand a given path to an absolute path. :type path: str :param path: The path to expand. :type path: str :type base_directory: str, optional :param base_directory: The base directory to use for expanding the path. Defaults to None. :type base_directory: str, optional

Returns:

The expanded absolute path.

Return type:

str

Raises:

None –

Note

This method handles both Unix and Windows paths and expands ‘~’ notation to the user’s home directory.

static expand_and_join(path1, path2)

Join two paths after expanding the first path.

Parameters:

  • path1 (str) – The first path to expand and join.

  • path2 (str) – The second path to join with the expanded first path.

Returns:

The joined path.

Return type:

str

classmethod from_toml(env, settings_path=None, section=None, **overrides)
Return type:

BaseWorker

static loop(*, poll_interval=None)

Run a long-lived service loop on this worker until signalled to stop.

The derived worker can implement a loop method accepting either zero arguments or a single stop_event argument. When the method signature accepts stop_event (keyword stop_event or should_stop), the worker implementation is responsible for honouring the event. Otherwise the base implementation repeatedly invokes the method and sleeps for the configured poll interval between calls. Returning False from the worker method requests termination of the loop.

Return type:

Dict[str, Any]

managed_pc_home_suffix: ClassVar[str] = 'MyApp'
managed_pc_path_fields: ClassVar[tuple[str, ...]] = ()
static normalize_dataset_path(data_path)

Normalise any dataset directory input so workers rely on consistent paths.

Return type:

str

prepare_output_dir(root, *, subdir='dataframe', attribute='data_out', clean=True)

Create (and optionally reset) a deterministic output directory.

Return type:

Path

setup_args(args, *, env=None, error=None, output_field=None, output_subdir='dataframe', output_attr='data_out', output_clean=True, output_parents_up=0)
Return type:

Any

setup_data_directories(*, source_path, target_path=None, target_subdir='dataframe', reset_target=False)

Prepare normalised input/output dataset paths without relying on worker args.

Returns a namespace with the resolved input path (input_path), the normalised string used by downstream readers (normalized_input), the output directory as a Path (output_path), and its normalised string representation (normalized_output). Optionally clears and recreates the output directory.

Return type:

SimpleNamespace

static start(worker_inst)

Invoke the concrete worker’s start hook once initialised.

stop()

Returns:

to_toml(settings_path=None, section=None, create_missing=True)
Return type:

None

verbose = 1
work_done(df=None)

Handles the post-processing of the DataFrame after work_pool execution.

Parameters:

df (pl.DataFrame, optional) – The Polars DataFrame to process. Defaults to None.

Raises:

ValueError – If an unsupported output format is specified.

Return type:

None

work_pool(x=None)

Processes a single task.

Parameters:

x (any, optional) – The task to process. Defaults to None.

Returns:

A Polars DataFrame with the processed results.

Return type:

pl.DataFrame

works(workers_plan, workers_plan_metadata)

Executes worker tasks based on the distribution tree.

Parameters:

  • workers_plan (any) – Distribution tree structure.

  • workers_plan_metadata (any) – Additional information about the workers.

Returns:

Execution time in seconds.

Return type:

float

mycode worker class diagram