Core

Registry of named callables (Registry + Command pattern).

The registry decouples “what to run” (a string name inside a JSON action list) from “how to run it” (a Python callable). Executors delegate name resolution to an ActionRegistry, which keeps look-up O(1) and lets plugins add commands at runtime without touching the executor class.

class automation_file.core.action_registry.ActionRegistry(initial=None)

Bases: object

Mapping of action name -> callable.

Parameters:

initial (Mapping[str, Command] | None)

property event_dict: dict[str, Callable[[...], Any]]

Backwards-compatible view used by older package_manager style code.

names()

Return type:

Iterable[str]

register(name, command)

Add or overwrite a command. Raises if command is not callable.

Parameters:

• name (str)

• command (Callable[[...], Any])

Return type:

None

register_many(mapping)

Register every name -> command pair in mapping.

Parameters:

mapping (Mapping[str, Callable[[...], Any]])

Return type:

None

resolve(name)

Parameters:

name (str)

Return type:

Callable[[…], Any] | None

unregister(name)

Parameters:

name (str)

Return type:

None

update(mapping)

Alias for register_many() (dict-compatible).

Parameters:

mapping (Mapping[str, Callable[[...], Any]])

Return type:

None

automation_file.core.action_registry.build_default_registry()

Return a registry pre-populated with every built-in FA_* action.

After the built-ins are registered, any third-party package advertising an automation_file.actions entry point is loaded so its commands land in the same registry. Plugins may override built-in names.

Return type:

ActionRegistry

Callback executor — runs a trigger, then a callback.

Implements the “do X then do Y” flow many automation JSON files want. The registry is shared with ActionExecutor, so adding a command to one adds it to the other.

class automation_file.core.callback_executor.CallbackExecutor(registry)

Bases: object

Invoke trigger(**kwargs) then callback(*args | **kwargs).

Parameters:

registry (ActionRegistry)

callback_function(trigger_function_name, callback_function, callback_function_param=None, callback_param_method='kwargs', **kwargs)

Parameters:

• trigger_function_name (str)

• callback_function (Callable[[...], Any])

• callback_function_param (Mapping[str, Any] | list[Any] | None)

• callback_param_method (str)

• kwargs (Any)

Return type:

Any

Dynamic plugin registration into an ActionRegistry.

PackageLoader imports an external package by name and registers every top-level function / class / builtin under the key "<package>_<member>".

class automation_file.core.package_loader.PackageLoader(registry)

Bases: object

Load packages lazily and register their public callables.

Parameters:

registry (ActionRegistry)

add_package_to_executor(package)

Register every function / class / builtin from package.

Returns the number of commands that were registered.

Parameters:

package (str)

Return type:

int

load(package)

Import package once and return the module (cached).

Parameters:

package (str)

Return type:

ModuleType | None

JSON persistence for action lists.

Reads/writes are serialised through a module-level lock so concurrent callers cannot interleave writes against the same file.

automation_file.core.json_store.read_action_json(json_file_path)

Return the parsed JSON content at json_file_path.

Parameters:

json_file_path (str)

Return type:

Any

automation_file.core.json_store.write_action_json(json_save_path, action_json)

Write action_json to json_save_path as pretty UTF-8 JSON.

Parameters:

• json_save_path (str)

• action_json (Any)

Return type:

None

Retry helper for transient network failures.

retry_on_transient is a small wrapper around exponential back-off. It is intentionally dependency-free so that modules which do not actually use requests or googleapiclient can import it without pulling those in.

automation_file.core.retry.retry_on_transient(max_attempts=3, backoff_base=0.5, backoff_cap=8.0, retriable=(<class 'ConnectionError'>, <class 'TimeoutError'>, <class 'OSError'>))

Return a decorator that retries retriable exceptions with back-off.

On the final failure raises RetryExhaustedException chained to the underlying error so callers can still inspect the cause.

Parameters:

• max_attempts (int)

• backoff_base (float)

• backoff_cap (float)

• retriable (tuple[type[BaseException], ...])

Return type:

Callable[[F], F]

Per-action quota enforcement.

Quota bundles a maximum byte size and maximum duration. Callers use Quota.check_size(bytes) before an I/O-heavy action and wrap the action in with quota.time_budget(label): to bound wall-clock time.

class automation_file.core.quota.Quota(max_bytes=0, max_seconds=0.0)

Bases: object

Bundle of per-action limits.

max_bytes <= 0 means no size cap; max_seconds <= 0 means no time cap. Defaults allow callers to share one Quota instance across many actions with fine-grained overrides at each call site.

Parameters:

• max_bytes (int)

• max_seconds (float)

check_size(nbytes, label='action')

Raise QuotaExceededException if nbytes exceeds the cap.

Parameters:

• nbytes (int)

• label (str)

Return type:

None

max_bytes: int = 0

max_seconds: float = 0.0

time_budget(label='action')

Context manager that raises if the enclosed block runs past the cap.

Parameters:

label (str)

Return type:

Iterator[None]

wraps(label, size_fn=None)

Return a decorator that enforces the time budget around func.

If size_fn is provided it is called with the function’s return value to derive a byte count for check_size().

Parameters:

label (str)

Token-bucket rate limiter.

RateLimiter refills at rate tokens/second up to a burst capacity. Callers acquire N tokens before issuing a protected call; when empty, the limiter either blocks (up to timeout) or raises RateLimitExceededException.

class automation_file.core.rate_limit.RateLimiter(rate, burst=None)

Bases: object

Thread-safe token bucket.

Parameters:

• rate (float)

• burst (float | None)

acquire(tokens=1.0, timeout=None)

Block until tokens are available.

Raises RateLimitExceededException if timeout elapses first. timeout=None waits indefinitely; timeout=0 fails immediately.

Parameters:

• tokens (float)

• timeout (float | None)

Return type:

None

property capacity: float

try_acquire(tokens=1.0)

Take tokens without blocking. Return True on success.

Parameters:

tokens (float)

Return type:

bool

wraps(tokens=1.0, timeout=None)

Return a decorator that acquires tokens before each call.

Parameters:

• tokens (float)

• timeout (float | None)

Return type:

Callable[[F], F]

Three-state circuit breaker.

States: CLOSED (normal), OPEN (short-circuit after failure_threshold consecutive failures), HALF_OPEN (trial one call after recovery_timeout seconds; one success closes, one failure re-opens). Failures are counted only for exceptions in retriable — internal errors surface as-is.

class automation_file.core.circuit_breaker.CircuitBreaker(failure_threshold=5, recovery_timeout=30.0, retriable=(<class 'Exception'>, ), name='circuit')

Bases: object

Open-close-half-open breaker.

failure_threshold — consecutive failures that trip the breaker. recovery_timeout — seconds spent in OPEN before transitioning to HALF_OPEN. retriable — exception types counted as failures; other exceptions pass through.

Parameters:

• failure_threshold (int)

• recovery_timeout (float)

• retriable (tuple[type[BaseException], ...])

• name (str)

call(func, *args, **kwargs)

Invoke func through the breaker.

Parameters:

• func (Callable[[...], Any])

• args (Any)

• kwargs (Any)

Return type:

Any

reset()

Force the breaker back to CLOSED, clearing failure count.

Return type:

None

property state: str

wraps()

Return a decorator that routes every call through call().

Return type:

Callable[[F], F]

Cross-platform advisory file lock.

Uses fcntl.flock on POSIX and msvcrt.locking on Windows, so two processes can serialise on a well-known lock path. Locks are exclusive (writer-style); shared locks are not supported because msvcrt cannot express them portably.

class automation_file.core.file_lock.FileLock(path, timeout=None)

Bases: object

Advisory exclusive lock on a sidecar lock file.

path is the lock file itself — typically <resource>.lock next to the protected resource. timeout is the maximum seconds to wait when acquiring; None waits indefinitely, 0 fails immediately.

Parameters:

• path (str | os.PathLike[str])

• timeout (float | None)

acquire()

Block until the lock is held. Raises LockTimeoutException on timeout.

Return type:

None

property is_held: bool

property path: Path

release()

Release the lock; idempotent.

Return type:

None

SQLite-backed named lock for multi-process / multi-host coordination.

Unlike automation_file.core.file_lock.FileLock which locks a single file descriptor, SQLiteLock persists named leases in a shared SQLite database. Any process that can open the database can participate. Leases carry an optional TTL so crashed owners eventually free the slot.

class automation_file.core.sqlite_lock.SQLiteLock(db_path, name, timeout=None, ttl=None)

Bases: object

Named lease stored in SQLite.

db_path is the SQLite file — callers sharing a lock must point at the same file. name is the lock identity. ttl (seconds) lets a crashed owner’s lease expire; None means the lease is held until explicit release. timeout bounds acquisition wait.

Parameters:

• db_path (str | os.PathLike[str])

• name (str)

• timeout (float | None)

• ttl (float | None)

acquire()

Block until the lease is granted; raise LockTimeoutException on timeout.

Return type:

None

property is_held: bool

property owner: str

refresh()

Extend the lease by ttl seconds. No-op when ttl is unset.

Return type:

None

release()

Release the lease; idempotent. Only the owning instance removes the row.

Return type:

None

Persistent SQLite-backed queue of action payloads.

Producers call ActionQueue.enqueue() to durably store a JSON action list; consumers pull with dequeue() (marking the row inflight) and finalise with ack() or nack(). The queue survives process restarts — all state lives in the SQLite file.

class automation_file.core.action_queue.ActionQueue(db_path)

Bases: object

Durable FIFO / priority queue for JSON action payloads.

Parameters:

db_path (str | os.PathLike[str])

ack(item_id)

Finalise a claimed row as processed.

Parameters:

item_id (int)

Return type:

None

dead_letters()

Return type:

list[QueueItem]

dequeue()

Claim the next ready row; returns None if the queue is empty.

Return type:

QueueItem | None

enqueue(action, priority=0, run_at=None)

Persist action for later dispatch. Returns the row id.

Parameters:

• action (list[Any] | dict[str, Any])

• priority (int)

• run_at (float | None)

Return type:

int

nack(item_id, *, requeue=True, reason='', delay=0.0)

Return a claimed row to the queue (requeue=True) or mark as dead.

Parameters:

• item_id (int)

• requeue (bool)

• reason (str)

• delay (float)

Return type:

None

purge()

Delete every row (ready / inflight / dead). Returns rows deleted.

Return type:

int

size(status='ready')

Parameters:

status (str)

Return type:

int

class automation_file.core.action_queue.QueueItem(id, action, attempts, enqueued_at)

Bases: object

A claimed queue row returned by ActionQueue.dequeue().

Parameters:

• id (int)

• action (list[Any] | dict[str, Any])

• attempts (int)

• enqueued_at (float)

action: list[Any] | dict[str, Any]

attempts: int

enqueued_at: float

id: int

SHA-256 content-addressable store.

ContentStore ingests files or byte blobs and keys them by the hex digest of their contents. A two-character fanout directory keeps any single directory small: <root>/ab/abcdef…. Identical inputs map to the same blob — callers get deduplication for free.

class automation_file.core.content_store.ContentStore(root)

Bases: object

Filesystem-backed CAS under root.

Parameters:

root (str | os.PathLike[str])

copy_to(digest, destination)

Copy the blob at digest into destination. Returns the path.

Parameters:

• digest (str)

• destination (str | PathLike[str])

Return type:

Path

delete(digest)

Remove a blob. Returns True when the blob existed.

Parameters:

digest (str)

Return type:

bool

exists(digest)

Parameters:

digest (str)

Return type:

bool

iter_digests()

Yield the digest of every stored blob.

Return type:

Iterator[str]

open(digest)

Open the stored blob for binary read.

Parameters:

digest (str)

Return type:

IO[bytes]

path_for(digest)

Parameters:

digest (str)

Return type:

Path

put(source)

Ingest the file at source and return its hex digest.

Parameters:

source (str | PathLike[str])

Return type:

str

put_bytes(data)

Ingest raw bytes and return the hex digest.

Parameters:

data (bytes)

Return type:

str

property root: Path

size()

Return the total number of stored blobs.

Return type:

int

Transfer progress + cancellation primitives.

Long-running transfers (HTTP downloads, S3 uploads/downloads, …) accept a named handle from the shared progress_registry. The registry keeps a ProgressReporter (bytes transferred, optional total) and a CancellationToken per name so the GUI or a JSON action can observe progress or cancel mid-flight.

Instrumentation is opt-in: callers pass progress_name="<label>" to enable tracking. When omitted, transfers run exactly as before with zero overhead beyond one attribute lookup.

class automation_file.core.progress.CancellationToken

Bases: object

Thread-safe boolean flag, pollable from worker threads.

cancel()

Return type:

None

property is_cancelled: bool

raise_if_cancelled()

Return type:

None

exception automation_file.core.progress.CancelledException

Bases: FileAutomationException

Raised when a cancellable operation is asked to stop mid-flight.

class automation_file.core.progress.ProgressRegistry

Bases: object

Named handles so JSON actions / the GUI can address ongoing transfers.

cancel(name)

Parameters:

name (str)

Return type:

bool

clear_finished()

Return type:

int

create(name, total=None)

Parameters:

• name (str)

• total (int | None)

Return type:

tuple[ProgressReporter, CancellationToken]

forget(name)

Parameters:

name (str)

Return type:

bool

list()

Return type:

list[dict[str, Any]]

lookup(name)

Parameters:

name (str)

Return type:

tuple[ProgressReporter, CancellationToken] | None

class automation_file.core.progress.ProgressReporter(name, total=None, transferred=0, status='running', started_at=<factory>, finished_at=None, _lock=<factory>)

Bases: object

Tracks bytes transferred for one named operation.

Parameters:

• name (str)

• total (int | None)

• transferred (int)

• status (str)

• started_at (float)

• finished_at (float | None)

• _lock (allocate_lock)

finish(status='done')

Parameters:

status (str)

Return type:

None

finished_at: float | None = None

property is_finished: bool

name: str

snapshot()

Return type:

dict[str, Any]

started_at: float

status: str = 'running'

total: int | None = None

transferred: int = 0

update(delta)

Parameters:

delta (int)

Return type:

None

automation_file.core.progress.progress_cancel(name)

Cancel the named transfer. Returns False if no such handle.

Parameters:

name (str)

Return type:

bool

automation_file.core.progress.progress_clear()

Drop every finished transfer from the registry.

Return type:

int

automation_file.core.progress.progress_list()

Snapshot of every registered transfer.

Return type:

list[dict[str, Any]]

automation_file.core.progress.register_progress_ops(registry)

Wire FA_progress_* actions into an ActionRegistry.

Parameters:

registry (Any)

Return type:

None

File checksum + integrity verification helpers.

Streaming hashes so multi-GB files don’t blow up memory. The hash algorithm is whatever hashlib.new() understands (sha256, sha1, md5, blake2b, …). file_checksum() returns the hex digest; verify_checksum() does a constant-time compare against the expected value so callers can’t leak timing on the check.

exception automation_file.core.checksum.ChecksumMismatchException

Bases: FileAutomationException

Raised when a computed digest does not match the expected value.

automation_file.core.checksum.file_checksum(path, algorithm='sha256', chunk_size=1048576)

Return the hex digest of path under algorithm.

Reads the file in chunk_size blocks so the memory cost is bounded regardless of file size. Raises FileNotExistsException when the path is missing and ValueError for unknown algorithms.

Parameters:

• path (str | PathLike[str])

• algorithm (str)

• chunk_size (int)

Return type:

str

automation_file.core.checksum.verify_checksum(path, expected, algorithm='sha256', chunk_size=1048576)

Return True iff path’s digest matches expected (case-insensitive).

Uses hmac.compare_digest() so the match check is constant-time.

Parameters:

• path (str | PathLike[str])

• expected (str)

• algorithm (str)

• chunk_size (int)

Return type:

bool

Directory-tree manifests — JSON snapshot of every file’s checksum.

A manifest is a simple JSON document recording every file under a root, its size, and a streaming digest (SHA-256 by default). Two operations:

write_manifest(root, manifest_path)  # snapshot now
verify_manifest(root, manifest_path) # verify the tree still matches

Use cases: release-artifact verification, backup integrity checks, detecting tampering on a sync target, or pre-flight checks before a rename / move.

The document shape is intentionally small and human-readable:

{
  "version": 1,
  "algorithm": "sha256",
  "root": "/abs/path/at/snapshot/time",
  "created_at": "2026-04-21T10:15:30+00:00",
  "files": {
    "a.txt":            {"size": 3,      "checksum": "..."},
    "nested/b.txt":     {"size": 1_024,  "checksum": "..."}
  }
}

Paths in the files mapping are POSIX-style (forward-slash) relative paths so the manifest round-trips across Windows and Unix.

exception automation_file.core.manifest.ManifestException

Bases: FileAutomationException

Raised for invalid manifest documents or unreadable manifest paths.

automation_file.core.manifest.verify_manifest(root, manifest_path)

Verify every file recorded in manifest_path still matches under root.

Returns a summary dict:

{
    "matched":  ["a.txt"],
    "missing":  ["gone.txt"],
    "modified": ["changed.txt"],
    "extra":    ["new.txt"],    # present under root, not in manifest
    "ok": False,
}

ok is True iff missing and modified are both empty (extras are reported but do not fail verification — mirror sync_dir’s default non-deleting posture).

Parameters:

• root (str | PathLike[str])

• manifest_path (str | PathLike[str])

Return type:

dict[str, Any]

automation_file.core.manifest.write_manifest(root, manifest_path, *, algorithm='sha256')

Write a manifest for every file under root. Returns the manifest dict.

Parameters:

• root (str | PathLike[str])

• manifest_path (str | PathLike[str])

• algorithm (str)

Return type:

dict[str, Any]

SQLite-backed audit log for executed actions.

AuditLog(db_path) opens (or creates) a single-table SQLite database and appends one row per action execution. Rows carry the timestamp, action name, a JSON-encoded snapshot of the payload, the result / error repr, and the duration in milliseconds.

Writes use a short-lived connection per call (check_same_thread=False semantics) so the log is safe to share between background worker threads and the scheduler. Readers call AuditLog.recent() to pull the most recent N rows.

The module deliberately avoids buffering / background queues: every row is persisted synchronously with an INSERT inside a with connect(..) so a crash at most loses the currently-executing action.

exception automation_file.core.audit.AuditException

Bases: FileAutomationException

Raised when the audit log cannot be opened or written.

class automation_file.core.audit.AuditLog(db_path)

Bases: object

Synchronous SQLite audit log.

Parameters:

db_path (str | Path)

count()

Return type:

int

purge(older_than_seconds)

Delete rows older than older_than_seconds and return the row count.

Parameters:

older_than_seconds (float)

Return type:

int

recent(limit=100)

Return the newest limit rows, newest first.

Parameters:

limit (int)

Return type:

list[dict[str, Any]]

record(action, payload, *, result=None, error=None, duration_ms=0.0)

Append a single audit row. Never raises — failures are logged only.

Parameters:

• action (str)

• payload (Any)

• result (Any)

• error (BaseException | None)

• duration_ms (float)

Return type:

None

Opt-in variable substitution for action list payloads.

When execute_action(..., substitute=True) is used, every string inside the action list is scanned for ${kind} / ${kind:arg} placeholders before dispatch. The following kinds are supported:

• ${env:NAME} — value of the NAME environment variable (empty when unset)

• ${date:FMT} — datetime.now().strftime(FMT); bare ${date} yields ISO

• ${uuid} — a fresh uuid.uuid4().hex

• ${cwd} — os.getcwd()

Unknown kinds raise SubstitutionException so typos surface loudly rather than leaking literal ${...} into paths.

exception automation_file.core.substitution.SubstitutionException

Bases: FileAutomationException

Raised when a ${...} reference names an unknown kind.

automation_file.core.substitution.substitute(payload)

Return a deep copy of payload with every ${...} expanded.

Parameters:

payload (object)

Return type:

object

Entry-point plugin discovery.

Third-party packages can register additional actions with automation_file without the library having to import them directly.

A plugin advertises itself in its pyproject.toml:

[project.entry-points."automation_file.actions"]
my_plugin = "my_plugin:register"

where register is a zero-argument callable returning a Mapping[str, Callable] — the same shape you would pass to automation_file.add_command_to_executor().

load_entry_point_plugins() is invoked by automation_file.core.action_registry.build_default_registry() so installed plugins populate every freshly-built registry automatically. Plugin failures are logged and swallowed — one broken plugin must not break the library for everyone else.

automation_file.core.plugins.load_entry_point_plugins(register)

Discover and register every automation_file.actions entry point.

register receives one {name: callable} mapping per plugin and is responsible for storing it (typically ActionRegistry.register_many()). Returns the number of plugins that registered successfully.

Parameters:

register (Callable[[Mapping[str, Callable[[...], Any]]], None])

Return type:

int

Secret provider abstraction.

A secret is a (name -> value) lookup. Three built-in providers compose via ChainedSecretProvider:

• EnvSecretProvider resolves from os.environ

• FileSecretProvider resolves from a directory of per-secret files (Docker / K8s secrets layout)

• ChainedSecretProvider tries a list in order

The purpose is to keep secrets out of the config file itself. Callers write references like ${env:SLACK_WEBHOOK_URL} inside automation_file.toml; resolve_secret_refs() walks the document and substitutes. Missing secrets raise SecretNotFoundException so a typo in a reference never silently becomes an empty string.

class automation_file.core.secrets.ChainedSecretProvider(providers)

Bases: SecretProvider

Try each child provider in order; return the first non-None value.

Providers are grouped by scheme — ${env:X} only consults EnvSecretProvider children, ${file:X} only consults FileSecretProvider children. Unknown schemes raise SecretNotFoundException at resolution time.

Parameters:

providers (list[SecretProvider])

get(name)

Return the secret’s value, or None if this provider has no entry.

Parameters:

name (str)

Return type:

str | None

resolve_one(scheme, name)

Parameters:

• scheme (str)

• name (str)

Return type:

str

scheme: str = '*'

class automation_file.core.secrets.EnvSecretProvider

Bases: SecretProvider

Read secrets from process environment variables.

get(name)

Return the secret’s value, or None if this provider has no entry.

Parameters:

name (str)

Return type:

str | None

scheme: str = 'env'

class automation_file.core.secrets.FileSecretProvider(root)

Bases: SecretProvider

Read secrets from <root>/<name> (e.g., /run/secrets/<name>).

Parameters:

root (str | os.PathLike[str])

get(name)

Return the secret’s value, or None if this provider has no entry.

Parameters:

name (str)

Return type:

str | None

scheme: str = 'file'

exception automation_file.core.secrets.SecretException

Bases: FileAutomationException

Base for secret-provider failures.

exception automation_file.core.secrets.SecretNotFoundException

Bases: SecretException

Raised when a ${provider:name} reference cannot be resolved.

class automation_file.core.secrets.SecretProvider

Bases: ABC

Contract for a read-only secret lookup.

abstractmethod get(name)

Return the secret’s value, or None if this provider has no entry.

Parameters:

name (str)

Return type:

str | None

scheme: str

automation_file.core.secrets.default_provider(file_root=None)

Return an env-first chain, optionally augmented with a file provider.

Parameters:

file_root (str | PathLike[str] | None)

Return type:

ChainedSecretProvider

automation_file.core.secrets.resolve_secret_refs(value, provider)

Walk value (dict/list/str) and substitute every ${scheme:name}.

Strings without references are returned unchanged; non-string scalars pass through untouched. Any unresolved reference raises SecretNotFoundException — callers should treat this as a hard configuration error rather than continuing with a hole in the config.

Parameters:

• value (Any)

• provider (ChainedSecretProvider)

Return type:

Any

Exception hierarchy for automation_file.

All custom exceptions inherit from FileAutomationException so callers can filter with a single except and still distinguish specific failures.

exception automation_file.exceptions.AddCommandException

Bases: FileAutomationException

Raised when a command registered into the executor is not callable.

exception automation_file.exceptions.ArchiveException

Bases: FileAutomationException

Raised when an archive format is unsupported or extraction fails.

exception automation_file.exceptions.ArgparseException

Bases: FileAutomationException

Raised when the CLI receives no actionable argument.

exception automation_file.exceptions.BoxException

Bases: FileAutomationException

Raised by the Box backend.

exception automation_file.exceptions.CASException

Bases: FileAutomationException

Raised by the content-addressable store on integrity / I/O failures.

exception automation_file.exceptions.CallbackExecutorException

Bases: FileAutomationException

Raised by CallbackExecutor for registration / dispatch failures.

exception automation_file.exceptions.CircuitOpenException

Bases: FileAutomationException

Raised when a circuit breaker is open and short-circuits the protected call.

exception automation_file.exceptions.DagException

Bases: FileAutomationException

Raised when a DAG action list has a cycle, unknown dep, or duplicate id.

exception automation_file.exceptions.DataOpsException

Bases: FileAutomationException

Raised by CSV / JSONL / YAML / Parquet helpers.

exception automation_file.exceptions.DiffException

Bases: FileAutomationException

Raised when diff computation or patch application fails.

exception automation_file.exceptions.DirNotExistsException

Bases: FileAutomationException

Raised when a required directory is missing.

exception automation_file.exceptions.ExecuteActionException

Bases: FileAutomationException

Raised by ActionExecutor when an action list cannot be run.

exception automation_file.exceptions.FileAutomationException

Bases: Exception

Root of the automation_file exception tree.

exception automation_file.exceptions.FileNotExistsException

Bases: FileAutomationException

Raised when a required source file is missing.

exception automation_file.exceptions.FsspecException

Bases: FileAutomationException

Raised by the fsspec bridge on missing dependency or backend failures.

exception automation_file.exceptions.JsonActionException

Bases: FileAutomationException

Raised when JSON action files cannot be read or written.

exception automation_file.exceptions.LockTimeoutException

Bases: FileAutomationException

Raised when a lock acquire waits past its timeout.

exception automation_file.exceptions.MCPServerException

Bases: FileAutomationException

Raised by the MCP server bridge when a tool invocation fails.

exception automation_file.exceptions.OneDriveException

Bases: FileAutomationException

Raised by the OneDrive (Microsoft Graph) backend.

exception automation_file.exceptions.PathTraversalException

Bases: FileAutomationException

Raised when a user-supplied path escapes the allowed root.

exception automation_file.exceptions.QueueException

Bases: FileAutomationException

Raised by the persistent action queue on storage / dispatch errors.

exception automation_file.exceptions.QuotaExceededException

Bases: FileAutomationException

Raised when an action exceeds a configured size or duration quota.

exception automation_file.exceptions.RateLimitExceededException

Bases: FileAutomationException

Raised when a rate-limited call cannot acquire a token in the allotted wait.

exception automation_file.exceptions.RetryExhaustedException

Bases: FileAutomationException

Raised when a @retry_on_transient wrapped call runs out of attempts.

exception automation_file.exceptions.SMBException

Bases: FileAutomationException

Raised by the SMB/CIFS client on connection / protocol failures.

exception automation_file.exceptions.TCPAuthException

Bases: FileAutomationException

Raised when a TCP client fails shared-secret authentication.

exception automation_file.exceptions.TemplateException

Bases: FileAutomationException

Raised when template rendering fails (missing engine, syntax, I/O).

exception automation_file.exceptions.TextOpsException

Bases: FileAutomationException

Raised by text / binary file helpers (split, merge, sed, encoding_convert).

exception automation_file.exceptions.TracingException

Bases: FileAutomationException

Raised when OpenTelemetry tracing setup cannot be completed.

exception automation_file.exceptions.UrlValidationException

Bases: FileAutomationException

Raised when a URL fails scheme / host validation (SSRF guard).

exception automation_file.exceptions.ValidationException

Bases: FileAutomationException

Raised when an action list fails pre-execution validation.

exception automation_file.exceptions.VersioningException

Bases: FileAutomationException

Raised by the versioning helpers on retention / I/O failures.

exception automation_file.exceptions.WebDAVException

Bases: FileAutomationException

Raised by the WebDAV client on transport / protocol failures.

exception automation_file.exceptions.ZipInputException

Bases: FileAutomationException

Raised when a zip helper receives an unsupported input type.

Module-level logger for automation_file.

A single file_automation_logger is exposed. It writes to FileAutomation.log in append mode and mirrors every record to stderr via a custom handler. The handler list is rebuilt only once, even if the module is reloaded, so tests can import this safely.