# Python automation + AI

Scripts, integrations and workers close to the Python data and AI ecosystem.

uv · Pydantic · HTTPX / provider SDK · pytest

[Complete guide (Markdown)](/stacks/python-scripts-workers/guide.md) · [Structured record (JSON)](/stacks/python-scripts-workers/record.json)

Original source review: 2026-10-10. Core guidance rechecked: 2026-10-13. [Check scope](/research-freshness.json). Dependency combinations and setup need project-specific qualification.

## Use cases

- Reusable automation scripts

- Scheduled imports and transformations

- Model/tool orchestration workers

## Choose it when

- The deliverable is a bounded task or worker rather than a web product

- Python SDKs or data libraries materially simplify integrations

## Consider another path when

- A one-off script is being expanded into an undocumented product backend

- Hard real-time deadlines or a native deployment requirement dominate

## Working practices

- Use scoped credentials, explicit timeouts and bounded concurrency for external calls.

- Validate model outputs and tool inputs; keep approval boundaries separate from model reasoning.

- Test with deterministic fake services and recorded fixtures, including partial failure and replay.

- Use a durable workflow engine only when restart recovery is required; make external effects idempotent.

## Tools by job

### Packaging and environments: uv + pyproject.toml + committed uv.lock

One project entry point for dependencies, Python selection, execution and reproducible sync.

Pin a compatible Python minor and resolve actual package versions at implementation; do not hand-edit uv.lock. Keep environments and caches in a local development workspace.

- [https://docs.astral.sh/uv/guides/projects/](https://docs.astral.sh/uv/guides/projects/)

### Lint and format: Ruff

Fast lint and formatting feedback with one configuration.

Ruff does not replace static typing, tests, authorization checks or runtime validation.

- [https://docs.astral.sh/ruff/installation/](https://docs.astral.sh/ruff/installation/)

### Static typing: mypy

Explicit checks on application boundaries with a conservative established checker.

Configure strictness and third-party stubs; unchecked Any and untyped functions reduce coverage. No checker performance comparison was run.

- [https://mypy.readthedocs.io/en/stable/getting_started.html](https://mypy.readthedocs.io/en/stable/getting_started.html)

- [https://docs.astral.sh/ty/type-checking/](https://docs.astral.sh/ty/type-checking/)

### Testing: pytest with deterministic fake external services and captured fixtures

Fast repeatable tests can verify parsing, permissions and retry decisions.

Add adversarial outputs and partial-failure cases for agents; do not confuse output snapshots with task success.

- [https://docs.pytest.org/en/stable/getting-started.html](https://docs.pytest.org/en/stable/getting-started.html)

### Validation: Pydantic at configuration, tool-input and model-output boundaries

Reject malformed external structures before effects.

Define strictness intentionally; a validated plan still needs permission and domain checks.

- [https://pydantic.dev/docs/validation/latest/concepts/strict_mode/](https://pydantic.dev/docs/validation/latest/concepts/strict_mode/)

### External I/O: Provider-supported Python SDK or HTTPX with explicit timeouts and bounded concurrency

Centralize transport settings, credentials and test seams.

Configure total task deadlines as well as network inactivity timeouts; avoid retry storms and duplicated writes; redact credentials from logs.

- [https://www.python-httpx.org/advanced/timeouts/](https://www.python-httpx.org/advanced/timeouts/)

### Persistence and migrations: No database for stateless scripts; PostgreSQL + SQLAlchemy/Alembic for shared durable business state

Avoid introducing infrastructure before state ownership requires it.

Files need atomic writes/checkpoints; database state does not automatically resume execution. Review migrations and use task-local sessions.

- [https://docs.sqlalchemy.org/en/21/orm/session_basics.html](https://docs.sqlalchemy.org/en/21/orm/session_basics.html)

- [https://alembic.sqlalchemy.org/en/latest/autogenerate.html](https://alembic.sqlalchemy.org/en/latest/autogenerate.html)

### Auth and credentials: Scoped service credentials injected by the execution platform

Workers generally need service authorization rather than user login UI.

Credential storage/provider is deployment-specific and unselected; keep secrets out of source, fixtures, task payloads and output.

- [https://docs.astral.sh/uv/guides/integration/docker/](https://docs.astral.sh/uv/guides/integration/docker/)

- [https://www.python-httpx.org/advanced/timeouts/](https://www.python-httpx.org/advanced/timeouts/)

### Long-running durable agents: Temporal Python SDK when recovery across restarts is required

Persisted workflow orchestration and separate worker processes suit retries, pauses and human waits.

Adds a service and operational cost. Keep external I/O in activities, make side effects idempotent, test replay and cancellation; no end-to-end integration was run.

- [https://docs.temporal.io/develop/python](https://docs.temporal.io/develop/python)

- [https://docs.celeryq.dev/en/stable/userguide/tasks.html](https://docs.celeryq.dev/en/stable/userguide/tasks.html)

### Execution and deployment: uv-managed package and explicit CLI entry point; separate managed worker process for persistent workers

Make inputs, exit codes and restart behavior reproducible.

Scheduler/platform is unselected; document non-overlap and checkpoint semantics. Do not add a web server just to run a job.

- [https://docs.astral.sh/uv/guides/projects/](https://docs.astral.sh/uv/guides/projects/)

- [https://docs.astral.sh/uv/guides/scripts/](https://docs.astral.sh/uv/guides/scripts/)

- [https://docs.astral.sh/uv/guides/integration/docker/](https://docs.astral.sh/uv/guides/integration/docker/)

### Application logging: Python logging with configured handlers

Shared standard logging captures application and dependency diagnostics

Configure logging at entry point, redact credentials and add request/job IDs, duration and failure context. Persist logs on the execution platform; metrics/tracing backend is a separate operational choice.

- [https://docs.python.org/3/library/logging.html](https://docs.python.org/3/library/logging.html)

### Bounded external-call retries: Tenacity only for retryable transient failures

Explicit stop/wait/filter rules support controlled retries

Install tenacity; set attempt/deadline limits, jitter and exception filters. The bare decorator retries forever without delay; do not stack retries with SDK/worker retries or repeat non-idempotent effects without keys.

- [https://tenacity.readthedocs.io/en/latest/](https://tenacity.readthedocs.io/en/latest/)

### Object storage: Boto3 S3 client when durable objects are required

Official SDK supports file transfer to S3

Install boto3 only for S3-backed work; use platform credentials, private keys and bounded transfers. Boto3 is synchronous: keep blocking transfers out of an async event loop; verify alternative provider compatibility.

- [https://docs.aws.amazon.com/boto3/latest/guide/s3-uploading-files.html](https://docs.aws.amazon.com/boto3/latest/guide/s3-uploading-files.html)

### HTTP timeouts: HTTPX client timeouts and connection limits

Connect/read/write/pool deadlines prevent unbounded remote waits

Configure shared client timeout and limits; retain overall operation deadline separately because read timeout measures inactivity between chunks. Close clients and redact request logs.

- [https://www.python-httpx.org/advanced/timeouts/](https://www.python-httpx.org/advanced/timeouts/)

Build with the stack’s strengths

## Architecture and developer workflow.

Recommendation: a maintained package owns validated inputs, a run/effect ledger, a CLI result contract and a bounded execution loop. A short task may run directly under a scheduler; independent queued jobs fit Celery when broker semantics are qualified; long waits and restart recovery may justify Temporal workflows with external effects in Activities. A prompt, validated model output or log is not proof of a completed effect.

Version scope: Python APIs: TaskGroup/asyncio.timeout require 3.11+; inspected /3 docs currently show 3.15.0. Use a supported minor compatible with selected SDKs and lock it; this does not recommend blindly upgrading to 3.15. Reviewed uv current guides, HTTPX current guides, Celery 5.6.3 task docs and Temporal Python guides. Resolve actual SDK versions in uv.lock before qualification. · Reviewed 2026-10-11

### Architecture boundaries

- Recommendation: separate parsing/validation, pure transformation, provider adapters and effects. Use deterministic fixtures/dry-run for decision logic, scoped injected credentials and stable operation IDs; do not place secrets or raw provider tokens in run payloads.

- Recommendation: reuse HTTPX clients with explicit connect/read/write/pool timeouts and connection limits; add a total task deadline and bounded TaskGroup/semaphore or worker-queue execution. Connection limits alone do not bound all queued coroutines.

- Recommendation: persist source cursor, input/schema version, effect intent and acknowledged result where restart matters. One-machine SQLite can own local state; shared workers need a chosen shared durable store and lease/ownership model. A checkpoint is not atomically committed with a remote email or payment.

- Recommendation: in Temporal, deterministic Workflow code orchestrates Activities, waits and retries; Activities perform provider calls and heartbeat progress. In Celery, configure acknowledgement/redelivery and retry policy to the task's replay semantics. Avoid layering unbounded SDK, decorator and worker retries.

### Tools and feedback

- Recommendation: inspect existing scheduler, dependencies, input formats and receipts. For maintained local work, uv init --package worker-app creates a package; pin a supported Python minor, add SDK/validation dependencies and commit pyproject.toml plus uv.lock in the maintained source repository.

- Recommendation: uv sync --locked, uv run ruff check ., uv run mypy src and uv run pytest require the corresponding dev tools configured/installed. Add fixtures for malformed provider/model output, permission failures, partial effects, ambiguous acknowledgements and resume.

- Recommendation: define [project.scripts] for the package CLI and structured JSON results plus explicit success/usage/failure statuses. uv run --locked <entrypoint> uses the locked project; scheduler start, overlap policy, termination and resume command are deployment-owned choices.

- Recommendation: for durable orchestration, uv add temporalio and implement Workflow/Activity/Worker components connected to a separately operated Temporal service. Replay representative histories with temporalio.worker.Replayer before workflow changes; test Activity heartbeat/cancellation and time-skipping integration flows.

- Recommendation: use uv sync --locked --no-dev inside a reviewed container build, then record the actual entry point, writable state path, credential scopes and dependency/image pins. No scheduler, broker or Temporal integration was executed here.

### Translations to avoid

- Avoid resuming by rerunning every input: a crash after remote success before checkpoint creates ambiguous effects and needs reconciliation or a provider idempotency key.

- Avoid synchronous SDK calls in async execution; use a documented async client or bounded offload, with explicit cancellation limits.

- Avoid swallowing CancelledError in TaskGroup work; cooperative cancellation depends on it being propagated after cleanup.

- Avoid describing Celery late acknowledgement or a Temporal Workflow as exactly-once external effects. Application/provider idempotency remains necessary.

- Avoid importing time/random/network behavior into deterministic Temporal Workflow code; move external I/O into Activities and replay-test changes.

[docs.astral.sh →](https://docs.astral.sh/uv/guides/projects/)
[docs.astral.sh →](https://docs.astral.sh/uv/guides/integration/docker/)
[docs.python.org →](https://docs.python.org/3/library/sqlite3.html)
[docs.python.org →](https://docs.python.org/3/library/os.html#os.replace)
[docs.python.org →](https://docs.python.org/3/library/asyncio-task.html)
[docs.python.org →](https://docs.python.org/3/library/asyncio-sync.html)
[www.python-httpx.org →](https://www.python-httpx.org/advanced/timeouts/)
[www.python-httpx.org →](https://www.python-httpx.org/advanced/resource-limits/)
[docs.celeryq.dev →](https://docs.celeryq.dev/en/stable/userguide/tasks.html)
[docs.temporal.io →](https://docs.temporal.io/develop/python/workflows/basics)
[docs.temporal.io →](https://docs.temporal.io/develop/python/activities/basics)
[docs.temporal.io →](https://docs.temporal.io/develop/python/activities/timeouts)
[docs.temporal.io →](https://docs.temporal.io/develop/python/best-practices/testing-suite)
[docs.temporal.io →](https://docs.temporal.io/activity-definition)

For an existing application

## Keep the contracts while you change the stack.

These are documented options and proposed checks. Inspect the actual source application before adopting a transition.

CLI and pipe contracts

### Inspect the source

Capture old script arguments, environment/config precedence, file paths, exit codes and scheduler invocation behavior.

### During coexistence

Recommendation: retain the old script as a thin entry point into an importable package while callers transition. Introduce structured output/version fields and an explicit noninteractive/dry-run mode without changing live side effects.

### Target design

Recommendation: uv package with pyproject.toml entry point, committed uv.lock and supported Python minor; JSON results on stdout, diagnostics on stderr, schema and status policy in the CLI documentation.

### Rollback

Recommendation: restore the previous package/environment and command while preserving versioned run/checkpoint files. Keep the old entry point until every scheduler/caller is migrated.

### Benefits and prerequisites

**An importable locked package makes repeated automation easier to test and invoke consistently.**
Requires: Inputs/entry point/status are defined; compatible Python and SDKs are pinned; scheduler uses the same package build.
Tradeoffs: A lockfile does not fix provider behavior or scheduler overlap; deploy-time environment/credentials remain required.

### Checks before transfer

- Run old scheduler command against the package adapter and compare input/output/status.

- Start from a clean locked environment and parse output without progress/log contamination.

- Verify dry-run performs zero provider writes and reports the intended actions.

[docs.astral.sh →](https://docs.astral.sh/uv/guides/projects/)
[docs.astral.sh →](https://docs.astral.sh/uv/guides/integration/docker/)

Data and schema

### Inspect the source

Identify source pagination/cursors, input revisions, item identifiers, partial outputs and acknowledged versus uncertain effects.

### During coexistence

Recommendation: introduce a run/effect ledger alongside the old logs and compare reconciliation in fixture mode. Keep one active execution owner; migrate old checkpoints with source-input/version provenance.

### Target design

Recommendation: transactional SQLite checkpoint/receipt state for one host or an operated shared database for multiple workers. For exported files, validate temporary output then os.replace on the same filesystem; qualify sync/permissions separately.

### Rollback

Recommendation: keep the ledger readable by the previous worker or convert it explicitly. Reconcile ambiguous remote outcomes before replay; reverting code cannot unsend a message or reverse a provider write.

### Benefits and prerequisites

**Durable progress records allow targeted restart and investigation instead of blind replay.**
Requires: Stable item/effect IDs; checkpoint transactions; single writer or tested shared ownership; provider lookup/idempotency for uncertain effects.
Tradeoffs: A file/database commit cannot atomically commit a remote side effect; os.replace atomicity does not establish power-loss durability or cross-filesystem replacement.

### Checks before transfer

- Crash before effect, after remote success and after checkpoint; verify each resume decision.

- Resume the same input and prove acknowledged effects are skipped or reconciled.

- Interrupt export and verify a complete prior/new artifact; restore ledger from backup and compare cursor ownership.

[docs.python.org →](https://docs.python.org/3/library/sqlite3.html)
[docs.python.org →](https://docs.python.org/3/library/os.html#os.replace)

Delivery and recovery

### Inspect the source

Preserve provider rate limits, retry budgets, total run deadline and cleanup/cancellation behavior; SDK inactivity timeout is not a whole-run deadline.

### During coexistence

Recommendation: retain sequential mode, add shared HTTPX clients and explicit capacity, and compare fixture results before enabling concurrency. Use a bounded queue for large inputs rather than creating one task for every item.

### Target design

Recommendation: TaskGroup plus asyncio.timeout for operation lifetime, semaphore/worker limits for admission, HTTPX network timeouts and connection limits. Keep sync SDK work outside the event loop; cancellation of offloaded work needs a separate policy.

### Rollback

Recommendation: return to sequential execution with the same receipt IDs and retry budget. Resolve uncertain effects from provider state before retrying; retain operation deadlines.

### Benefits and prerequisites

**Bounded concurrency and total deadlines prevent unlimited waiting work and reduce abandoned I/O.**
Requires: Supported Python 3.11+; every provider adapter honors limits; aggregate retry budget and cancellation cleanup are configured.
Tradeoffs: Read timeout measures inactivity between chunks; cancelling a coroutine does not prove a server write was cancelled; throughput gains are unmeasured.

### Checks before transfer

- Slow streaming chunks beyond the total deadline while below read timeout; the whole operation still stops.

- Overload input and verify active plus queued work remain bounded.

- Cancel one task during provider I/O; TaskGroup cleanup closes clients and reports any ambiguous write.

[docs.python.org →](https://docs.python.org/3/library/asyncio-task.html)
[docs.python.org →](https://docs.python.org/3/library/asyncio-sync.html)
[www.python-httpx.org →](https://www.python-httpx.org/advanced/timeouts/)
[www.python-httpx.org →](https://www.python-httpx.org/advanced/resource-limits/)

Jobs and schedules: Independent jobs with Celery

### Inspect the source

Capture independent task payload versions, broker durability, acknowledgement timing, retries, dead-letter handling and scheduler overlap.

### During coexistence

Recommendation: keep the old scheduler for unconverted tasks while routing selected task kinds into a qualified Celery queue. One scheduler/consumer owns each kind, and old backlog states remain accounted for.

### Target design

Recommendation: Celery for independent jobs, with idempotent task bodies, bounded worker concurrency, explicit retry/backoff and intentionally selected acknowledgement/worker-loss settings. Keep existing brokers when their guarantees satisfy the contract.

### Rollback

Recommendation: stop new producers, retain compatible workers for accepted tasks, and reroute only reconciled unstarted work. A scheduler rollback must not create a second schedule for the same task kind.

### Benefits and prerequisites

**A qualified queue permits independent worker restart and retry for bounded jobs.**
Requires: Broker persistence/redelivery and payload mapping are tested; task effects are replay-safe; acknowledgement and worker-loss policy are deliberately configured.
Tradeoffs: Late acknowledgement alone does not guarantee redelivery for every child-process termination; Celery worker settings and broker behavior matter; duplicate execution remains possible.

### Checks before transfer

- Terminate the worker during a task and verify actual redelivery for the selected settings.

- Lose acknowledgement after an external effect; retries must not duplicate the logical effect.

- Drain or transform ready/delayed/retry/failed work before retiring the old scheduler/consumer.

[docs.celeryq.dev →](https://docs.celeryq.dev/en/stable/userguide/tasks.html)

Jobs and schedules: Recoverable multi-step workflows with Temporal

### Inspect the source

Capture long-lived operation IDs, timer/human-wait behavior, provider effects and history/checkpoint schema before adopting durable orchestration.

### During coexistence

Recommendation: keep old workflows running under their existing owner; start only newly routed operation IDs in Temporal. Keep provider calls in Activities and preserve effect IDs across retries. Do not duplicate a live run as a shadow test.

### Target design

Recommendation: Temporal Python Workflows/Activities with specified timeouts, heartbeats, replay tests and compatible worker deployments. Temporal service/visibility/history retention and operational ownership are prerequisites, not Python package installation side effects.

### Rollback

Recommendation: stop routing new IDs to the new workflow version and keep compatible workers servicing existing histories. Replay-test the rollback code; if incompatible, complete/version old executions rather than abandoning their state.

### Benefits and prerequisites

**Persisted workflow history can resume orchestration across worker restarts and represent long waits.**
Requires: Temporal service and workers are operated; Workflow code is deterministic; Activity effects are idempotent/reconcilable; compatible histories pass replay.
Tradeoffs: Adds a service and history/versioning obligations; Activity retries may repeat effects; orchestration recovery is not automatic compensation of external changes.

### Checks before transfer

- Replay representative open and completed histories after code changes; fail promotion on nondeterminism.

- Restart workers during timers and Activities; verify orchestration resumes and heartbeat/checkpoint behavior.

- Crash after provider success before Activity completion; retry reconciles the stable effect ID.

[docs.temporal.io →](https://docs.temporal.io/develop/python/workflows/basics)
[docs.temporal.io →](https://docs.temporal.io/develop/python/activities/basics)
[docs.temporal.io →](https://docs.temporal.io/develop/python/activities/timeouts)
[docs.temporal.io →](https://docs.temporal.io/develop/python/best-practices/testing-suite)
[docs.temporal.io →](https://docs.temporal.io/activity-definition)

[Read the migration workflow →](/rewrite/guide.md)

## Setup

- For maintained work, create a local source project: uv init worker-app; cd worker-app; select/pin compatible Python; uv add httpx pydantic pydantic-settings

- uv add --dev ruff mypy pytest; define an importable worker_app package and a pyproject.toml CLI entry point.

- Implement explicit inputs, a dry-run or fixture mode, bounded external calls, structured result/error output, and credentials injected by the environment.

- For a genuinely standalone file, the alternative is uv init --script task.py then uv add --script task.py httpx pydantic; run via uv run task.py. Promote reused logic into a package.

- Only when shared persistent state is required: add sqlalchemy, psycopg and alembic, then configure reviewed migrations. For durable orchestration: uv add temporalio and implement workflows/activities plus a worker connected to a separately provisioned Temporal service.

- Package the locked environment for the chosen scheduler or worker platform. Record startup, cancellation, retry, checkpoint and failure-resume behavior; verify fixture execution before live access.

## Verification

Run these in the configured application; they are instructions, not recorded passes.

```
uv sync --locked
uv run ruff check .
uv run ruff format --check .
uv run mypy src
uv run pytest
uv run python -m compileall -q src
```

## Open questions

- Which provider SDKs and minimum credential scopes are required?

- Are runs bounded, scheduled, queued or long-lived with human waits?

- Which effects are safe to retry, and what are run deadlines and cost budgets?

- Who owns execution history, alerts, secrets and recovery?

## Sources

- [uv projects](https://docs.astral.sh/uv/guides/projects/)

- [Ruff installation](https://docs.astral.sh/ruff/installation/)

- [mypy getting started](https://mypy.readthedocs.io/en/stable/getting_started.html)

- [ty type checking](https://docs.astral.sh/ty/type-checking/)

- [pytest getting started](https://docs.pytest.org/en/stable/getting-started.html)

- [Pydantic strict mode](https://pydantic.dev/docs/validation/latest/concepts/strict_mode/)

- [HTTPX timeouts](https://www.python-httpx.org/advanced/timeouts/)

- [SQLAlchemy session basics](https://docs.sqlalchemy.org/en/21/orm/session_basics.html)

- [Alembic autogeneration](https://alembic.sqlalchemy.org/en/latest/autogenerate.html)

- [uv Docker](https://docs.astral.sh/uv/guides/integration/docker/)

- [Temporal Python SDK](https://docs.temporal.io/develop/python)

- [Celery tasks](https://docs.celeryq.dev/en/stable/userguide/tasks.html)

- [uv scripts](https://docs.astral.sh/uv/guides/scripts/)

- [Python logging](https://docs.python.org/3/library/logging.html)

- [Tenacity documentation](https://tenacity.readthedocs.io/en/latest/)

- [Boto3 uploads](https://docs.aws.amazon.com/boto3/latest/guide/s3-uploading-files.html)

---

Canonical page: https://stack.menu/stacks/python-scripts-workers/