blocks implementation#

status: implemented. this doc preserved as reference for API call sequences and database schema.

api call sequence#

When user calls block.save("name"):

1. GET /block_types/slug/{slug} → 404 if not found
2. POST /block_types/ → create type (or PATCH /block_types/{id} if exists)
3. GET /block_schemas/checksum/{checksum} → 404 if not found
4. POST /block_schemas/ → create schema
5. POST /block_documents/ → create document
   • if 409 conflict (name exists): GET + PATCH to update

When user calls Block.load("name"):

1. GET /block_types/slug/{slug}/block_documents/name/{name} → return document with nested schema/type

database tables#

block_type (
    id TEXT PRIMARY KEY,
    created, updated,
    name TEXT NOT NULL,
    slug TEXT NOT NULL UNIQUE,
    logo_url, documentation_url, description, code_example TEXT,
    is_protected INTEGER DEFAULT 0
)

block_schema (
    id TEXT PRIMARY KEY,
    created, updated,
    checksum TEXT NOT NULL,
    fields TEXT DEFAULT '{}',      -- JSON schema
    capabilities TEXT DEFAULT '[]', -- JSON array
    version TEXT DEFAULT '1',
    block_type_id TEXT FK,
    UNIQUE(checksum, version)
)

block_document (
    id TEXT PRIMARY KEY,
    created, updated,
    name TEXT,
    data TEXT DEFAULT '{}',        -- JSON (encrypted in python, plain for us)
    is_anonymous INTEGER DEFAULT 0,
    block_type_id TEXT FK,
    block_type_name TEXT,          -- denormalized
    block_schema_id TEXT FK,
    UNIQUE(block_type_id, name)
)

implementation phases#

phase 1: save() support (minimum viable)#

• add tables to schema
• db/block_types.zig - insert, getBySlug, update
• db/block_schemas.zig - insert, getByChecksum
• db/block_documents.zig - insert, getById, update
• api/block_types.zig:
  • GET /block_types/slug/{slug}
  • POST /block_types/
  • PATCH /block_types/{id}
• api/block_schemas.zig:
  • GET /block_schemas/checksum/{checksum}
  • POST /block_schemas/
• api/block_documents.zig:
  • POST /block_documents/
  • PATCH /block_documents/{id}
  • GET /block_documents/{id}

phase 2: load() support#

• GET /block_types/slug/{slug}/block_documents/name/{name}

phase 3: filter endpoints#

• POST /block_types/filter
• POST /block_schemas/filter
• POST /block_documents/filter

phase 4: nested blocks (if needed)#

• block_schema_reference table
• block_document_reference table
• recursive document hydration

test script#

from prefect.blocks.system import Secret

# save
secret = Secret(value="my-secret-value")
secret.save("test-secret")

# load
loaded = Secret.load("test-secret")
print(loaded.get())

response formats#

See prefect source for exact JSON shapes:

• src/prefect/client/schemas/responses.py
• src/prefect/server/schemas/core.py