Great — here’s a minimal, end-to-end example for Prefect 3.x that deploys a flow via Python using local code (no Docker) to a local Prefect server, then runs it with a local process worker. Assumptions: - You’re on Prefect 3.x - You want to store code locally (no container image), and run on a local process worker - You’re running against a local Prefect server 1) Start the local Prefect server In one terminal:

prefect server start

If your client isn’t already pointed at the local server, set the API URL in another terminal:

export PREFECT_API_URL="<http://127.0.0.1:4200/api>"
# or persist it
prefect config set PREFECT_API_URL="<http://127.0.0.1:4200/api>"

2) Create a process work pool and start a worker

prefect work-pool create "local-process" --type process
prefect worker start --pool "local-process"

3) Create your flow and a Python deploy script Project structure:

your_project/
  flows.py
  deploy.py

flows.py:

from prefect import flow, get_run_logger

@flow
def my_flow(name: str = "world"):
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"Hello, {name}!")

deploy.py:

from pathlib import Path
from prefect import flow

project_root = Path(__file__).parent

deployment_id = flow.from_source(
    source=project_root,                 # local code directory
    entrypoint="flows.py:my_flow",       # file:function
).deploy(
    name="local",                        # deployment name
    work_pool_name="local-process",      # process work pool
    tags=["local"],
    parameters={"name": "Prefect"},
)

print(f"Created deployment with id: {deployment_id}")

Then run:

python deploy.py

4) Trigger a run of your deployment With your worker running:

prefect deployment run "my_flow/local" --watch

Notes: - Passing a local path to

flow.from_source(source=...)

automatically uses local code storage; no Docker image is built or pushed. - The

entrypoint

uses the

file.py:function_name

format relative to

source

. - For simple local development, code changes in your

source

directory are picked up at runtime by the process worker on the same machine; redeploy only when you change deployment metadata (e.g., work pool, schedule, parameters, tags, etc.). References: - Deploy via Python: Deploy via Python - Run flows in local processes: Run flows in local processes - Local Server CLI: Server CLI If you want me to tailor this to your exact project layout or add a schedule (interval/cron), share your structure and I’ll adjust the code.

Short take: your client is sending a deployment payload that includes a git-derived version_info, and the server you’re talking to rejects that field — hence 422 extra_forbidden at body.version_info. Prefect collects git metadata from your local repo to populate deployment versioning even if you aren’t “using git storage,” which is why git shows up here. The most common cause is a version mismatch between the Prefect client you’re using to deploy and the Prefect server you’re hitting. Why this happens - In Prefect 3.x, deployments include versioning metadata. The client can infer git info (commit, branch, etc.) from the local repo when you deploy. - The server’s deployment schema is strict about what fields it accepts. If the server version doesn’t expect clients to send version_info, it rejects it as an extra field. - This typically occurs when the client and server are on different major/minor versions. What to do 1) Align versions - Upgrade Prefect on both the deploy environment and Server A to the same Prefect 3.x version, then redeploy. - Verify with:

prefect version

Run that on: - the machine where you run deploy.py or

prefect deploy

- Server A (where Prefect server runs) 2) Re-run the deployment - Once versions match, your deploy should succeed; the server will handle versioning metadata correctly. A couple of related notes - You have a pull step that sets the working directory to

/mnt/serverA/code/dev

, but your worker is on Server B. That path likely won’t exist on B. For local-code runs on a remote worker, make sure the worker can access the code: - Put the code on B at a known path and point to that path, or - Use a shared/network path, or - Use a repo/object storage the worker can pull from at runtime. See: Store flow code - The git metadata in your payload is just Prefect’s versioning inference from your local repo. It doesn’t mean you must use git-based storage. Can you share: - The output of

prefect version

on your deploy machine, on Server A, and on Server B? - How you’re deploying (Python

flow.from_source(...).deploy(...)

vs

prefect deploy

)? - Whether

/mnt/serverA/code/dev

is reachable from the worker on B? References: - Deployment versioning overview: Deployment versioning - Running flows in local processes (no Docker): Run flows in local processes - Store flow code patterns: Store flow code

This error means your Prefect server is configured to use PostgreSQL, but the target database hasn’t been created yet. The

prefect server database upgrade -y

command only runs schema migrations; it does not create the database itself. Since the connection is via asyncpg, your connection URL is pointing to Postgres (likely to a DB named

prefect

) that doesn’t exist yet. How to fix it Option A — You want PostgreSQL 1) Create the database, then run the upgrade:

# One way (from a psql shell connected to the server):
CREATE DATABASE prefect;

# Or from your shell:
createdb -h <pg_host> -U <pg_user> prefect

Then:

prefect server database upgrade -y

You can also just start the server (it runs migrations on start by default), but the DB must still exist:

prefect server start

Option B — You want SQLite for local dev Point Prefect to SQLite instead of Postgres, then upgrade/start:

export PREFECT_SERVER_DATABASE_CONNECTION_URL="sqlite+aiosqlite:///$(pwd)/prefect.db"
prefect server database upgrade -y
# or just:
prefect server start

Quick diagnostics - See what DB URL Prefect is using:

prefect config view | grep PREFECT_SERVER_DATABASE_CONNECTION_URL -n

If you see

postgresql+asyncpg://.../prefect

, that’s why asyncpg is in play and why the DB must be created first. Refs - Server CLI: Server CLI - Settings reference (DB connection URL): Settings reference If you share your current

PREFECT_SERVER_DATABASE_CONNECTION_URL

, I can suggest the exact create command or a correct SQLite URL for your environment.

You’re right to be surprised — but that setting only applies to events, not logs. - PREFECT_SERVER_EVENTS_RETENTION_PERIOD controls the retention of the events subsystem. - Flow/task run logs are stored in a separate logs table and are not affected by the events retention setting. Ways to control logs growth A) Prune logs now (PostgreSQL) If your server uses Postgres, a quick cleanup is:

-- delete logs older than 90 days
DELETE FROM log
WHERE "timestamp" < NOW() - INTERVAL '90 days';

-- reclaim space (optional, may lock)
VACUUM (VERBOSE, ANALYZE) log;
-- or full vacuum during a maintenance window:
-- VACUUM FULL log;

Tip: ensure an index on "timestamp" to speed up deletes:

CREATE INDEX IF NOT EXISTS idx_log_timestamp ON log ("timestamp");

B) Prune logs via API (automation option) You can run a periodic job that deletes logs older than N days. The server exposes a DELETE /logs endpoint (with filters) and per-run deletion endpoints. A simple approach is to list flow runs older than N days and delete their logs:

import asyncio
from datetime import datetime, timedelta, timezone
from prefect import get_client

async def prune_logs_older_than(days: int):
    cutoff = datetime.now(timezone.utc) - timedelta(days=days)
    async with get_client() as client:
        # page through old flow runs
        offset = 0
        limit = 200
        while True:
            runs = await client.read_flow_runs(
                limit=limit,
                offset=offset,
                # filter by start/end time server-side if you wish;
                # for simplicity we filter client-side here
            )
            if not runs:
                break
            old_runs = [r for r in runs if (r.end_time or r.start_time or r.created) and
                        ((r.end_time or r.start_time or r.created) < cutoff)]
            for r in old_runs:
                # call delete endpoint for each run’s logs
                await client._client.request(
                    "DELETE",
                    f"/flow_runs/{r.id}/logs",
                )
            if len(runs) < limit:
                break
            offset += limit

asyncio.run(prune_logs_older_than(90))

Run this weekly via cron. C) Reduce what gets stored going forward - Lower the level of logs sent to the API: - Set

PREFECT_LOGGING_TO_API_LEVEL=WARNING

(or ERROR) to reduce volume. - Or disable sending logs to the API entirely and keep logs in stdout/your own sink: -

PREFECT_LOGGING_TO_API_ENABLED=false

Why events retention didn’t help - Events and logs are different data models and retention paths.

PREFECT_SERVER_EVENTS_RETENTION_PERIOD="P90D"

prunes events, not the logs table. If you want, I can tailor a pruning script to use the global DELETE /logs filter (one request vs. per-run) for your exact server version. Are you running Postgres or SQLite, and which Prefect version is on your server?