Tracking 

update_run_status(run_id: str, status: str, error: str | None = None) → None[source]

Update workflow run status.

Parameters:

run_id (str) – Run ID
status (str) – New status
error (str | None) – Optional error message

Raises:

TaskException – If run not found
StorageException – If storage operation fails
TaskStateError – If status transition is invalid

Return type:

None

create_task(node_id: str, input_data: dict[str, Any] | None = None, metadata: dict[str, Any] | None = None, run_id: str = 'test-run-id', node_type: str = 'default-node-type', dependencies: list[str] | None = None, started_at: datetime | None = None) → TaskRun[source]

Create a new task.

Parameters:

node_id (str) – Node ID in the workflow
input_data (dict[str, Any] | None) – Input data for the task
metadata (dict[str, Any] | None) – Additional metadata
run_id (str) – Associated run ID (defaults to “test-run-id” for backward compatibility)
node_type (str) – Type of node (defaults to “default-node-type” for backward compatibility)
dependencies (list[str] | None) – List of task IDs this task depends on
started_at (datetime | None) – When the task started

Returns:

TaskRun instance

Raises:

TaskException – If task creation fails
StorageException – If storage operation fails

Return type:

TaskRun

update_task_status(task_id: str, status: TaskStatus, result: dict[str, Any] | None = None, error: str | None = None, ended_at: datetime | None = None, metadata: dict[str, Any] | None = None) → None[source]

Update task status.

Parameters:

task_id (str) – Task ID
status (TaskStatus) – New status
result (dict[str, Any] | None) – Task result
error (str | None) – Error message
ended_at (datetime | None) – When the task ended
metadata (dict[str, Any] | None) – Additional metadata

Raises:

TaskException – If task not found
StorageException – If storage operation fails
TaskStateError – If status transition is invalid

Return type:

None

get_run(run_id: str) → WorkflowRun | None[source]

Get workflow run by ID.

Parameters:: run_id (str) – Run ID
Returns:: WorkflowRun instance or None
Raises:: StorageException – If storage operation fails
Return type:: WorkflowRun | None

get_task(task_id: str) → TaskRun | None[source]

Get task by ID.

Parameters:: task_id (str) – Task ID
Returns:: TaskRun instance or None
Raises:: StorageException – If storage operation fails
Return type:: TaskRun | None

list_runs(workflow_name: str | None = None, status: str | None = None, limit: int | None = None) → list[RunSummary][source]

List workflow runs.

Parameters:

workflow_name (str | None) – Filter by workflow name
status (str | None) – Filter by status
limit (int | None) – Maximum number of runs to return

Returns:

List of run summaries

Raises:

StorageException – If storage operation fails

Return type:

list[RunSummary]

list_tasks(run_id: str, node_id: str | None = None, status: TaskStatus | None = None) → list[TaskSummary][source]

List tasks for a run.

Parameters:

run_id (str) – Run ID
node_id (str | None) – Filter by node ID
status (TaskStatus | None) – Filter by status

Returns:

List of task summaries

Raises:

TaskException – If run_id is not provided
StorageException – If storage operation fails

Return type:

list[TaskSummary]

get_run_summary(run_id: str) → RunSummary | None[source]

Get summary for a specific run.

Parameters:: run_id (str) – Run ID
Returns:: RunSummary or None
Raises:: StorageException – If storage operation fails
Return type:: RunSummary | None

clear_cache() → None[source]

Clear in-memory caches.

Return type:: None

complete_task(task_id: str, output_data: dict[str, Any] | None = None) → None[source]

Complete a task successfully.

Parameters:

task_id (str) – Task ID
output_data (dict[str, Any] | None) – Output data for the task

Raises:

TaskException – If task not found
StorageException – If storage operation fails

Return type:

None

fail_task(task_id: str, error_message: str) → None[source]

Mark a task as failed.

Parameters:

task_id (str) – Task ID
error_message (str) – Error message

Raises:

TaskException – If task not found
StorageException – If storage operation fails

Return type:

None

cancel_task(task_id: str, reason: str) → None[source]

Cancel a task.

Parameters:

task_id (str) – Task ID
reason (str) – Cancellation reason

Raises:

TaskException – If task not found
StorageException – If storage operation fails

Return type:

None

retry_task(task_id: str) → TaskRun[source]

Create a new task as a retry of an existing task.

Parameters:

task_id (str) – Original task ID

Returns:

New task instance

Raises:

TaskException – If task not found
StorageException – If storage operation fails

Return type:

TaskRun

delete_task(task_id: str) → None[source]

Delete a task.

Parameters:

task_id (str) – Task ID

Raises:

TaskException – If task not found
StorageException – If storage operation fails

Return type:

None

get_tasks_by_status(status: TaskStatus) → list[TaskRun][source]

Get tasks by status.

Parameters:: status (TaskStatus) – Status to filter by
Returns:: List of matching tasks
Raises:: StorageException – If storage operation fails
Return type:: list[TaskRun]

get_tasks_by_node(node_id: str) → list[TaskRun][source]

Get tasks by node ID.

Parameters:: node_id (str) – Node ID to filter by
Returns:: List of matching tasks
Raises:: StorageException – If storage operation fails
Return type:: list[TaskRun]

get_task_history(task_id: str) → list[TaskRun][source]

Get task history (original task and all retries).

Parameters:

task_id (str) – Task ID

Returns:

List of tasks in order (original first, latest retry last)

Raises:

TaskException – If task not found
StorageException – If storage operation fails

Return type:

list[TaskRun]

get_tasks_by_timerange(start_time: datetime, end_time: datetime) → list[TaskRun][source]

Get tasks created between start_time and end_time.

Parameters:

start_time (datetime) – Start of time range
end_time (datetime) – End of time range

Returns:

List of matching tasks

Raises:

StorageException – If storage operation fails

Return type:

list[TaskRun]

get_task_statistics() → dict[str, Any][source]

Get task statistics.

Returns:

total_tasks: Total number of tasks
by_status: Count of tasks by status
by_node: Count of tasks by node ID

Return type:

Dictionary with statistics

Raises:

StorageException – If storage operation fails

cleanup_old_tasks(days: int = 30) → int[source]

Delete tasks older than specified days.

Parameters:: days (int) – Age in days
Returns:: Number of tasks deleted
Raises:: StorageException – If storage operation fails
Return type:: int

update_task_metrics(task_id: str, metrics: TaskMetrics) → None[source]

Update task metrics.

Parameters:

task_id (str) – Task ID
metrics (TaskMetrics) – Metrics to update

Raises:

TaskException – If task not found
StorageException – If storage operation fails

Return type:

None

get_running_tasks() → list[TaskRun][source]

Get all currently running tasks.

Returns:: List of running tasks
Raises:: StorageException – If storage operation fails
Return type:: list[TaskRun]

get_task_dependencies(task_id: str) → list[TaskRun][source]

Get tasks that are dependencies for the given task.

Parameters:

task_id (str) – Task ID

Returns:

List of dependency tasks

Raises:

TaskException – If task not found
StorageException – If storage operation fails

Return type:

list[TaskRun]

save_task(task: TaskRun) → None[source]

Save a task to storage.

This is a convenience method that directly saves a task instance to storage. For new tasks, prefer using create_task() instead.

Parameters:: task (TaskRun) – TaskRun instance to save
Raises:: StorageException – If storage operation fails
Return type:: None

get_run_tasks(run_id: str) → list[TaskRun][source]

Get all tasks for a specific run.

Parameters:: run_id (str) – Run ID to get tasks for
Returns:: List of tasks in the run
Return type:: list[TaskRun]

set_search_attributes(run_id: str, attributes: dict[str, Any]) → None[source]

Set (upsert) search attributes on a workflow run.

Parameters:

run_id (str) – Workflow run ID.
attributes (dict[str, Any]) – Key-value pairs to store.

Raises:

TaskException – If run_id is empty.
StorageException – If storage operation fails.

Return type:

None

search_runs(filters: dict[str, Any], order_by: str = 'created_at DESC', limit: int = 100, offset: int = 0) → list[dict][source]

Search workflow runs by attribute filters.

Parameters:

filters (dict[str, Any]) – Attribute name-value pairs to match.
order_by (str) – Order by clause (column + direction).
limit (int) – Maximum results.
offset (int) – Results to skip.

Returns:

List of workflow run dicts matching all filters.

Raises:

StorageException – If storage operation fails.

Return type:

list[dict]

get_execution_audit_trail(run_id: str) → list[dict][source]

Get a comprehensive execution audit trail for a run.

Combines task records with timing and audit events into a chronological list of everything that happened during the run.

Parameters:

run_id (str) – Workflow run ID.

Returns:

Chronological list of audit trail entries.

Raises:

TaskException – If run_id is empty.
StorageException – If storage operation fails.

Return type:

list[dict]

get_workflow_tasks(workflow_id: str) → list[TaskRun][source]

Get all tasks for a workflow.

This is a compatibility method that returns all tasks across all runs for a workflow. In practice, tasks are tracked per run, not per workflow.

Parameters:: workflow_id (str) – Workflow ID (used to filter runs)
Returns:: List of all TaskRun objects for the workflow
Raises:: StorageException – If storage operation fails
Return type:: list[TaskRun]

Basic Usage:

from kailash.tracking import TaskManager
from kailash import Workflow
from kailash.workflow.runner import WorkflowRunner

# Create task manager
task_manager = TaskManager()

# Use with workflow runner
workflow = Workflow("my_workflow")
runner = WorkflowRunner(workflow, task_manager)

# Execute and track
results = runner.run()

# Access tracking data
run_id = results["run_id"]
run_info = task_manager.get_run(run_id)
print(f"Execution took: {run_info.duration}s")

Tracking Models 

WorkflowRun 

Represents a complete workflow execution.

class kailash.tracking.models.WorkflowRun(*, run_id: str = <factory>, workflow_name: str, status: str = 'running', started_at: datetime = <factory>, ended_at: datetime | None = None, tasks: list[str] = <factory>, metadata: dict[str, ~typing.Any]=<factory>, error: str | None = None)[source]

Bases: BaseModel

Model for a workflow execution run.

Parameters:

run_id (str)
workflow_name (str)
status (str)
started_at (datetime)
ended_at (datetime | None)
tasks (list[str])
metadata (dict[str, Any])
error (str | None)

run_id: str

workflow_name: str

status: str

started_at: datetime

ended_at: datetime | None

tasks: list[str]

metadata: dict[str, Any]

error: str | None

classmethod validate_workflow_name(v)[source]: Validate workflow name is not empty.

classmethod validate_status(v)[source]: Validate status is valid.

update_status(status: str, error: str | None = None) → None[source]

Update run status.

Parameters:

status (str) – New status
error (str | None) – Error message (for failed runs)

Raises:

TaskStateError – If state transition is invalid

Return type:

None

add_task(task_id: str) → None[source]

Add a task to this run.

Parameters:: task_id (str) – Task ID to add
Raises:: TaskException – If task_id is invalid
Return type:: None

get_duration() → float | None[source]

Get run duration in seconds.

Returns:: Duration in seconds, or None if not completed
Return type:: float | None

to_dict() → dict[str, Any][source]

Convert to dictionary representation.

Return type:: dict[str, Any]

__copy__() → Self

Returns a shallow copy of the model.

Return type:: Self

__deepcopy__(memo: dict[int, Any] | None = None) → Self

Returns a deep copy of the model.

Parameters:: memo (dict[int, Any] | None)
Return type:: Self

classmethod __get_pydantic_json_schema__(core_schema: CoreSchema, handler: GetJsonSchemaHandler, /) → JsonSchemaValue

Hook into generating the model’s JSON schema.

Parameters:

core_schema (CoreSchema) – A pydantic-core CoreSchema. You can ignore this argument and call the handler with a new CoreSchema, wrap this CoreSchema ({‘type’: ‘nullable’, ‘schema’: current_schema}), or just call the handler with the original schema.
handler (GetJsonSchemaHandler) – Call into Pydantic’s internal JSON schema generation. This will raise a pydantic.errors.PydanticInvalidForJsonSchema if JSON schema generation fails. Since this gets called by BaseModel.model_json_schema you can override the schema_generator argument to that function to change JSON schema generation globally for a type.

Returns:

A JSON schema, as a Python object.

Return type:

JsonSchemaValue

__init__(**data: Any) → None

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.

Parameters:: data (Any)
Return type:: None

__iter__() → Generator[tuple[str, Any], None, None]

So dict(model) works.

Return type:: Generator[tuple[str, Any], None, None]

__pretty__(fmt: Callable[[Any], Any], **kwargs: Any) → Generator[Any]

Used by devtools (https://python-devtools.helpmanual.io/) to pretty print objects.

Parameters:

fmt (Callable[[Any], Any])
kwargs (Any)

Return type:

Generator[Any]

classmethod __pydantic_init_subclass__(**kwargs: Any) → None

This is intended to behave just like __init_subclass__, but is called by ModelMetaclass only after basic class initialization is complete. In particular, attributes like model_fields will be present when this is called, but forward annotations are not guaranteed to be resolved yet, meaning that creating an instance of the class may fail.

This is necessary because __init_subclass__ will always be called by type.__new__, and it would require a prohibitively large refactor to the ModelMetaclass to ensure that type.__new__ was called in such a manner that the class would already be sufficiently initialized.

This will receive the same kwargs that would be passed to the standard __init_subclass__, namely, any kwargs passed to the class definition that aren’t used internally by Pydantic.

Parameters:: **kwargs (Any) – Any keyword arguments passed to the class definition that aren’t used internally by Pydantic.
Return type:: None

Note

You may want to override [__pydantic_on_complete__()][pydantic.main.BaseModel.__pydantic_on_complete__] instead, which is called once the class and its fields are fully initialized and ready for validation.

classmethod __pydantic_on_complete__() → None

This is called once the class and its fields are fully initialized and ready to be used.

This typically happens when the class is created (just before [__pydantic_init_subclass__()][pydantic.main.BaseModel.__pydantic_init_subclass__] is called on the superclass), except when forward annotations are used that could not immediately be resolved. In that case, it will be called later, when the model is rebuilt automatically or explicitly using [model_rebuild()][pydantic.main.BaseModel.model_rebuild].

Return type:: None

__repr_name__() → str

Name of the instance’s class, used in __repr__.

Return type:: str

__repr_recursion__(object: Any) → str

Returns the string representation of a recursive object.

Parameters:: object (Any)
Return type:: str

__rich_repr__() → RichReprResult

Used by Rich (https://rich.readthedocs.io/en/stable/pretty.html) to pretty print objects.

Return type:: RichReprResult

classmethod construct(_fields_set: set[str] | None = None, **values: Any) → Self

Parameters:

_fields_set (set[str] | None)
values (Any)

Return type:

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 (AbstractSetIntStr | MappingIntStrAny | None) – Optional set or mapping specifying which fields to include in the copied model.
exclude (AbstractSetIntStr | MappingIntStrAny | None) – Optional set or mapping specifying which fields to exclude in the copied model.
update (Dict[str, Any] | None) – 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.

Returns:

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

Return type:

Self

Parameters:

include (set[int] | set[str] | Mapping[int, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | Mapping[str, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | None)
exclude (set[int] | set[str] | Mapping[int, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | Mapping[str, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | None)
by_alias (bool)
exclude_unset (bool)
exclude_defaults (bool)
exclude_none (bool)

Return type:

classmethod from_orm(obj: Any) → Self

Parameters:: obj (Any)
Return type:: Self

Parameters:

include (set[int] | set[str] | Mapping[int, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | Mapping[str, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | None)
exclude (set[int] | set[str] | Mapping[int, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | Mapping[str, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | None)
by_alias (bool)
exclude_unset (bool)
exclude_defaults (bool)
exclude_none (bool)
encoder (Callable[[Any], Any] | None)
models_as_dict (bool)
dumps_kwargs (Any)

Return type:

model_computed_fields = {}

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

classmethod model_construct(_fields_set: set[str] | None = None, **values: Any) → Self

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.

Returns:

A new instance of the Model class with validated data.

Return type:

model_copy(*, update: Mapping[str, Any] | None = None, deep: bool = False) → Self

!!! 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.

Returns:

New model instance.

Return type:

!!! 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 (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 (set[int] | set[str] | Mapping[int, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | Mapping[str, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | None) – A set of fields to include in the output.
exclude (set[int] | set[str] | Mapping[int, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | Mapping[str, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, 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 (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 (Callable[[Any], Any] | None) – 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.
polymorphic_serialization (bool | None) – Whether to use model and dataclass polymorphic serialization for this call.

Returns:

A dictionary representation of the model.

Return type:

dict[str, Any]

!!! 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 (set[int] | set[str] | Mapping[int, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | Mapping[str, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | None) – Field(s) to include in the JSON output.
exclude (set[int] | set[str] | Mapping[int, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | Mapping[str, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, 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 (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 (Callable[[Any], Any] | None) – 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.
polymorphic_serialization (bool | None) – Whether to use model and dataclass polymorphic serialization for this call.

Returns:

A JSON string representation of the model.

Return type:

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 = {'ended_at': FieldInfo(annotation=Union[datetime, NoneType], required=False, default=None), 'error': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'metadata': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'run_id': FieldInfo(annotation=str, required=False, default_factory=<lambda>), 'started_at': FieldInfo(annotation=datetime, required=False, default_factory=<lambda>), 'status': FieldInfo(annotation=str, required=False, default='running', description='Run status'), 'tasks': FieldInfo(annotation=list[str], required=False, default_factory=list, description='Task IDs'), 'workflow_name': FieldInfo(annotation=str, required=True, description='Name of the workflow')}

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_parametrized_name(params: tuple[type[Any], ...]) → str

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.
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.
Return type:: str

model_post_init(context: Any, /) → None

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.

Parameters:: context (Any)
Return type:: None

classmethod model_rebuild(*, force: bool = False, raise_errors: bool = True, _parent_namespace_depth: int = 2, _types_namespace: MappingNamespace | None = None) → bool | 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 (MappingNamespace | None) – The types namespace, defaults to 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.

Return type:

bool | None

Validate a pydantic model instance.

Parameters:

obj (Any) – The object to validate.
strict (bool | None) – Whether to enforce types strictly.
extra (Literal['allow', 'ignore', 'forbid'] | None) – 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.

Returns:

The validated model instance.

Return type:

!!! 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 (Literal['allow', 'ignore', 'forbid'] | None) – 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.

Returns:

The validated Pydantic model.

Raises:

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

Return type:

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 (Literal['allow', 'ignore', 'forbid'] | None) – 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.

Returns:

The validated Pydantic model.

Return type:

classmethod parse_file(path: str | Path, *, content_type: str | None = None, encoding: str = 'utf8', proto: DeprecatedParseProtocol | None = None, allow_pickle: bool = False) → Self

Parameters:

path (str | Path)
content_type (str | None)
encoding (str)
proto (DeprecatedParseProtocol | None)
allow_pickle (bool)

Return type:

Self

classmethod parse_obj(obj: Any) → Self

Parameters:: obj (Any)
Return type:: Self

classmethod parse_raw(b: str | bytes, *, content_type: str | None = None, encoding: str = 'utf8', proto: DeprecatedParseProtocol | None = None, allow_pickle: bool = False) → Self

Parameters:

b (str | bytes)
content_type (str | None)
encoding (str)
proto (DeprecatedParseProtocol | None)
allow_pickle (bool)

Return type:

Self

classmethod schema(by_alias: bool = True, ref_template: str = '#/$defs/{model}') → Dict[str, Any]

Parameters:

by_alias (bool)
ref_template (str)

Return type:

classmethod schema_json(*, by_alias: bool = True, ref_template: str = '#/$defs/{model}', **dumps_kwargs: Any) → str

Parameters:

by_alias (bool)
ref_template (str)
dumps_kwargs (Any)

Return type:

classmethod update_forward_refs(**localns: Any) → None

Parameters:: localns (Any)
Return type:: None

classmethod validate(value: Any) → Self

Parameters:: value (Any)
Return type:: Self

Attributes:

run_id: Unique identifier for the run
workflow_id: Identifier of the executed workflow
status: Current status (pending, running, completed, failed)
start_time: Execution start timestamp
end_time: Execution end timestamp
duration: Total execution time
metadata: Additional run metadata

Example Usage:

# Get run information
run = task_manager.get_run(run_id)

print(f"Run ID: {run.run_id}")
print(f"Status: {run.status}")
print(f"Duration: {run.duration}s")
print(f"Nodes executed: {len(run.tasks)}")

Task 

Represents individual node execution within a workflow run.

kailash.tracking.models.Task: alias of TaskRun

Attributes:

task_id: Unique task identifier
run_id: Parent workflow run ID
node_id: ID of the executed node
status: Task status (pending, running, completed, failed, skipped)
start_time: Task start timestamp
end_time: Task end timestamp
duration: Task execution time
error: Error information if failed
metrics: Performance metrics

Example Usage:

# Get task details
tasks = task_manager.get_tasks(run_id)

for task in tasks:
    print(f"Node: {task.node_id}")
    print(f"Status: {task.status}")
    print(f"Duration: {task.duration}s")
    if task.error:
        print(f"Error: {task.error}")

TaskMetrics 

Performance and resource metrics for task execution.

class kailash.tracking.models.TaskMetrics(*, duration: float | None = 0.0, memory_usage: float | None = 0.0, memory_usage_mb: float | None = 0.0, cpu_usage: float | None = 0.0, custom_metrics: dict[str, ~typing.Any]=<factory>)[source]

Bases: BaseModel

Metrics for task execution.

Parameters:

duration (float | None)
memory_usage (float | None)
memory_usage_mb (float | None)
cpu_usage (float | None)
custom_metrics (dict[str, Any])

duration: float | None

memory_usage: float | None

memory_usage_mb: float | None

cpu_usage: float | None

custom_metrics: dict[str, Any]

__init__(**data)[source]: Initialize metrics with unified memory field handling.

classmethod validate_positive_metrics(v)[source]: Validate metric values are positive.

to_dict() → dict[str, Any][source]

Convert metrics to dictionary representation.

Return type:: dict[str, Any]

classmethod from_dict(data: dict[str, Any]) → TaskMetrics[source]

Create metrics from dictionary representation.

Parameters:: data (dict[str, Any])
Return type:: TaskMetrics

__copy__() → Self

Returns a shallow copy of the model.

Return type:: Self

__deepcopy__(memo: dict[int, Any] | None = None) → Self

Returns a deep copy of the model.

Parameters:: memo (dict[int, Any] | None)
Return type:: Self

classmethod __get_pydantic_json_schema__(core_schema: CoreSchema, handler: GetJsonSchemaHandler, /) → JsonSchemaValue

Hook into generating the model’s JSON schema.

Parameters:

core_schema (CoreSchema) – A pydantic-core CoreSchema. You can ignore this argument and call the handler with a new CoreSchema, wrap this CoreSchema ({‘type’: ‘nullable’, ‘schema’: current_schema}), or just call the handler with the original schema.
handler (GetJsonSchemaHandler) – Call into Pydantic’s internal JSON schema generation. This will raise a pydantic.errors.PydanticInvalidForJsonSchema if JSON schema generation fails. Since this gets called by BaseModel.model_json_schema you can override the schema_generator argument to that function to change JSON schema generation globally for a type.

Returns:

A JSON schema, as a Python object.

Return type:

JsonSchemaValue

__iter__() → Generator[tuple[str, Any], None, None]

So dict(model) works.

Return type:: Generator[tuple[str, Any], None, None]

__pretty__(fmt: Callable[[Any], Any], **kwargs: Any) → Generator[Any]

Used by devtools (https://python-devtools.helpmanual.io/) to pretty print objects.

Parameters:

fmt (Callable[[Any], Any])
kwargs (Any)

Return type:

Generator[Any]

classmethod __pydantic_init_subclass__(**kwargs: Any) → None

This is intended to behave just like __init_subclass__, but is called by ModelMetaclass only after basic class initialization is complete. In particular, attributes like model_fields will be present when this is called, but forward annotations are not guaranteed to be resolved yet, meaning that creating an instance of the class may fail.

This is necessary because __init_subclass__ will always be called by type.__new__, and it would require a prohibitively large refactor to the ModelMetaclass to ensure that type.__new__ was called in such a manner that the class would already be sufficiently initialized.

This will receive the same kwargs that would be passed to the standard __init_subclass__, namely, any kwargs passed to the class definition that aren’t used internally by Pydantic.

Parameters:: **kwargs (Any) – Any keyword arguments passed to the class definition that aren’t used internally by Pydantic.
Return type:: None

Note

You may want to override [__pydantic_on_complete__()][pydantic.main.BaseModel.__pydantic_on_complete__] instead, which is called once the class and its fields are fully initialized and ready for validation.

classmethod __pydantic_on_complete__() → None

This is called once the class and its fields are fully initialized and ready to be used.

This typically happens when the class is created (just before [__pydantic_init_subclass__()][pydantic.main.BaseModel.__pydantic_init_subclass__] is called on the superclass), except when forward annotations are used that could not immediately be resolved. In that case, it will be called later, when the model is rebuilt automatically or explicitly using [model_rebuild()][pydantic.main.BaseModel.model_rebuild].

Return type:: None

__repr_name__() → str

Name of the instance’s class, used in __repr__.

Return type:: str

__repr_recursion__(object: Any) → str

Returns the string representation of a recursive object.

Parameters:: object (Any)
Return type:: str

__rich_repr__() → RichReprResult

Used by Rich (https://rich.readthedocs.io/en/stable/pretty.html) to pretty print objects.

Return type:: RichReprResult

classmethod construct(_fields_set: set[str] | None = None, **values: Any) → Self

Parameters:

_fields_set (set[str] | None)
values (Any)

Return type:

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 (AbstractSetIntStr | MappingIntStrAny | None) – Optional set or mapping specifying which fields to include in the copied model.
exclude (AbstractSetIntStr | MappingIntStrAny | None) – Optional set or mapping specifying which fields to exclude in the copied model.
update (Dict[str, Any] | None) – 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.

Returns:

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

Return type:

Self

Parameters:

include (set[int] | set[str] | Mapping[int, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | Mapping[str, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | None)
exclude (set[int] | set[str] | Mapping[int, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | Mapping[str, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | None)
by_alias (bool)
exclude_unset (bool)
exclude_defaults (bool)
exclude_none (bool)

Return type:

classmethod from_orm(obj: Any) → Self

Parameters:: obj (Any)
Return type:: Self

Parameters:

include (set[int] | set[str] | Mapping[int, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | Mapping[str, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | None)
exclude (set[int] | set[str] | Mapping[int, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | Mapping[str, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | None)
by_alias (bool)
exclude_unset (bool)
exclude_defaults (bool)
exclude_none (bool)
encoder (Callable[[Any], Any] | None)
models_as_dict (bool)
dumps_kwargs (Any)

Return type:

model_computed_fields = {}

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

classmethod model_construct(_fields_set: set[str] | None = None, **values: Any) → Self

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.

Returns:

A new instance of the Model class with validated data.

Return type:

model_copy(*, update: Mapping[str, Any] | None = None, deep: bool = False) → Self

!!! 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.

Returns:

New model instance.

Return type:

!!! 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 (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 (set[int] | set[str] | Mapping[int, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | Mapping[str, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | None) – A set of fields to include in the output.
exclude (set[int] | set[str] | Mapping[int, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | Mapping[str, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, 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 (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 (Callable[[Any], Any] | None) – 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.
polymorphic_serialization (bool | None) – Whether to use model and dataclass polymorphic serialization for this call.

Returns:

A dictionary representation of the model.

Return type:

dict[str, Any]

!!! 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 (set[int] | set[str] | Mapping[int, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | Mapping[str, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | None) – Field(s) to include in the JSON output.
exclude (set[int] | set[str] | Mapping[int, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | Mapping[str, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, 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 (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 (Callable[[Any], Any] | None) – 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.
polymorphic_serialization (bool | None) – Whether to use model and dataclass polymorphic serialization for this call.

Returns:

A JSON string representation of the model.

Return type:

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 = {'cpu_usage': FieldInfo(annotation=Union[float, NoneType], required=False, default=0.0), 'custom_metrics': FieldInfo(annotation=dict[str, Any], required=False, default_factory=dict), 'duration': FieldInfo(annotation=Union[float, NoneType], required=False, default=0.0), 'memory_usage': FieldInfo(annotation=Union[float, NoneType], required=False, default=0.0), 'memory_usage_mb': FieldInfo(annotation=Union[float, NoneType], required=False, default=0.0)}

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_parametrized_name(params: tuple[type[Any], ...]) → str

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.
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.
Return type:: str

model_post_init(context: Any, /) → None

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.

Parameters:: context (Any)
Return type:: None

classmethod model_rebuild(*, force: bool = False, raise_errors: bool = True, _parent_namespace_depth: int = 2, _types_namespace: MappingNamespace | None = None) → bool | 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 (MappingNamespace | None) – The types namespace, defaults to 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.

Return type:

bool | None

Validate a pydantic model instance.

Parameters:

obj (Any) – The object to validate.
strict (bool | None) – Whether to enforce types strictly.
extra (Literal['allow', 'ignore', 'forbid'] | None) – 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.

Returns:

The validated model instance.

Return type:

!!! 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 (Literal['allow', 'ignore', 'forbid'] | None) – 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.

Returns:

The validated Pydantic model.

Raises:

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

Return type:

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 (Literal['allow', 'ignore', 'forbid'] | None) – 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.

Returns:

The validated Pydantic model.

Return type:

classmethod parse_file(path: str | Path, *, content_type: str | None = None, encoding: str = 'utf8', proto: DeprecatedParseProtocol | None = None, allow_pickle: bool = False) → Self

Parameters:

path (str | Path)
content_type (str | None)
encoding (str)
proto (DeprecatedParseProtocol | None)
allow_pickle (bool)

Return type:

Self

classmethod parse_obj(obj: Any) → Self

Parameters:: obj (Any)
Return type:: Self

classmethod parse_raw(b: str | bytes, *, content_type: str | None = None, encoding: str = 'utf8', proto: DeprecatedParseProtocol | None = None, allow_pickle: bool = False) → Self

Parameters:

b (str | bytes)
content_type (str | None)
encoding (str)
proto (DeprecatedParseProtocol | None)
allow_pickle (bool)

Return type:

Self

classmethod schema(by_alias: bool = True, ref_template: str = '#/$defs/{model}') → Dict[str, Any]

Parameters:

by_alias (bool)
ref_template (str)

Return type:

classmethod schema_json(*, by_alias: bool = True, ref_template: str = '#/$defs/{model}', **dumps_kwargs: Any) → str

Parameters:

by_alias (bool)
ref_template (str)
dumps_kwargs (Any)

Return type: