lacing

lacing — interval annotation system.

Standoff, interval-keyed annotations with rational time, ELAN tier stereotypes, Allen’s interval algebra, and a MutableMapping facade.

Quick start:

>>> from lacing import RationalTime, TimeInterval, Annotation, MemoryStore
>>> # Load a TextGrid, query overlaps, save as WebVTT — see misc/docs/.

Read CLAUDE.md and misc/docs/Lacing Development Roadmap.md for the full story. .claude/skills/ contains the rules.

class lacing.AllenRelation(value)

The thirteen Allen relations.

Symbols match Allen (1983); inverse pairs end in i.

inverse() → AllenRelation

The inverse relation.

class lacing.Annotation(*, id: UUID, tier: str, reference: MediaRef | NodeRef | AnnotationRef, body: dict, body_schema_uri: Annotated[str, _PydanticGeneralMetadata(pattern='^annot://schema/[a-z0-9-]+/v\\d+$')], provenance: Provenance, confidence: Annotated[float | None, Ge(ge=0.0), Le(le=1.0)] = None)

The single annotation envelope. body is typed by body_schema_uri.

property interval: TimeInterval | None

the reference’s interval, if any.

Type:

Convenience

model_config = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class lacing.AnnotationRef(*, kind: Literal['annotation'] = 'annotation', target_id: UUID, interval: TimeInterval | None = None)

Reference to another annotation (for discussion threads, review, derivations).

model_config = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class lacing.Artifact(*, asset_id: Annotated[str, MinLen(min_length=64), MaxLen(max_length=64), _PydanticGeneralMetadata(pattern='^[0-9a-f]{64}$')], kind: Literal['image', 'video', 'audio', 'json', 'text', 'binary'], path: Path | None = None, url: str | None = None, bytes_size: Annotated[int, Ge(ge=0)], duration_s: Annotated[float | None, Ge(ge=0)] = None, mime: str | None = None, provenance: Provenance, cost_usd: Annotated[float | None, Ge(ge=0)] = None, producer_call_id: str | None = None)

A content-addressed generated file with provenance.

asset_id is the SHA-256 hex digest of the artifact’s bytes. Two artifacts with the same asset_id are byte-identical regardless of where they live — so caches keyed on asset_id are safe across machines and re-runs.

provenance reuses lacing.Provenance so the lineage chain (was_derived_from, was_generated_by) is the same for artifacts and annotations. An annotation referencing an artifact does so via MediaRef(asset_id=artifact.asset_id, …).

classmethod from_bytes(data: bytes, *, kind: Literal['image', 'video', 'audio', 'json', 'text', 'binary'], was_generated_by: str, was_attributed_to: str, path: Path | str | None = None, url: str | None = None, was_derived_from: tuple = (), activity: str = 'create', generated_at_time: RationalTime | None = None, duration_s: float | None = None, mime: str | None = None, cost_usd: float | None = None, producer_call_id: str | None = None) → Artifact

Create an Artifact from in-memory bytes.

classmethod from_path(path: Path | str, *, kind: Literal['image', 'video', 'audio', 'json', 'text', 'binary'], was_generated_by: str, was_attributed_to: str, was_derived_from: tuple = (), activity: str = 'create', generated_at_time: RationalTime | None = None, duration_s: float | None = None, mime: str | None = None, cost_usd: float | None = None, producer_call_id: str | None = None) → Artifact

Create an Artifact from a local file. Hashes the file’s bytes.

model_config = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

to_media_ref(interval) → MediaRef

Return a lacing.MediaRef pointing at this artifact.

Use this to attach an annotation to a region of the artifact: MediaRef(asset_id=artifact.asset_id, interval=…).

class lacing.ArtifactStore(catalog: MutableMapping[str, BaseModel], blobs: MutableMapping[str, bytes] | None = None)

Facade over an artifact catalog and an optional blobs store.

The object is a MutableMapping[str, record] over the catalog — store[artifact_id], iteration, len, get, clear and the rest of the mapping surface all act on artifact metadata records. The heavier byte operations (put_blob(), get_blob(), has_blob()) are rich methods that are deliberately not squeezed into the mapping protocol.

Parameters:

• catalog – Injected id -> record store. Records are pydantic models (lacing.Artifact by default, but any BaseModel works — the store does not inspect the record’s shape).

• blobs – Injected content_hash -> bytes store, or None for a catalog-only store (Stage-1 metadata persistence). Blob methods raise / no-op when it is None.

Construct one with in_memory() or from_directory() rather than wiring the backing stores by hand, unless you are injecting a custom backend.

blob_path(content_hash: str) → Path | None

Return the local filesystem path of the blob, or None.

The store’s blob backend is opaque (any MutableMapping), but some backends — notably the filesystem-backed dol.Files produced by from_directory() — store each blob as one file under a known root directory. This method exposes that path when available, so a caller (e.g. a FastAPI route serving video) can hand the OS the file descriptor and let it answer HTTP Range requests directly. It returns None for:

• blob stores without a rootdir attribute (e.g. plain dict, object-store backends — callers should fall back to iter_blob());

• blobs that are not present.

Callers must treat None as the cue to use the streaming read path, not as an error.

classmethod from_directory(root: Path | str, *, record_type: type[BaseModel] = <class 'lacing.artifact.Artifact'>) → ArtifactStore

An ArtifactStore persisted under root.

Lays out two subdirectories: catalog/ (one <id>.json file per record) and blobs/ (one file per content hash). Both are dol filesystem stores, so the same facade works unchanged over any other dol backend (object storage, etc.) when injected directly.

Parameters:

• root – Directory to hold the store. Created if missing.

• record_type – The pydantic model the catalog deserializes JSON into. Defaults to Artifact; callers with their own record schema pass their model here.

get_blob(content_hash: str) → bytes | None

Return the bytes for content_hash, or None if absent.

has_blob(content_hash: str) → bool

Whether the blob store holds content_hash.

classmethod in_memory() → ArtifactStore

An ArtifactStore backed entirely by in-memory dicts.

For tests, scratch work, and as the trivial reference backend. Nothing persists across processes.

index() → dict[str, BaseModel]

Return the whole catalog as a plain dict (e.g. for UI hydration).

iter_blob(content_hash: str, *, chunk_size: int = 65536) → Iterator[bytes]

Yield the blob’s bytes in chunk_size chunks.

The streaming counterpart to get_blob() — what an HTTP response body iterates over when serving a large blob without holding it all in process memory. The default implementation reads the whole blob via get_blob() and re-chunks it; a filesystem-backed store can be swapped for a true streaming reader without changing this API.

Raises:

KeyError – no blob exists for content_hash.

put_blob(data: bytes) → str

Store data content-addressed; return its content hash.

Idempotent: identical bytes always map to the same hash and overwrite an identical blob.

Raises:

RuntimeError – no blob store is configured.

put_blob_stream(chunks: Iterable[bytes]) → str

Stream chunks content-addressed; return their SHA-256 hash.

The streaming-friendly counterpart to put_blob() — callers hand in an iterable (e.g. requests.Response.iter_content) instead of materializing the whole bytestring upfront. The hash is computed on the fly. The default implementation still buffers the bytes in the store’s own memory before writing; that is adequate for files up to a few hundred megabytes. A future filesystem-aware optimization (write-to-tempfile + atomic rename) is an internal change that does not touch this API.

Raises:

RuntimeError – no blob store is configured.

save(artifact_id: str, record: BaseModel, *, data: bytes | None = None) → str | None

Persist one artifact, optionally with its bytes.

Writes the blob (if data is given) before the catalog row, so a crash never leaves the catalog pointing at missing bytes. Idempotent on artifact_id — and, for the blob, on content — so retries are safe.

Parameters:

• artifact_id – The stable string id this artifact is filed under.

• record – The metadata record to store in the catalog.

• data – Optional raw bytes. When given, they are stored content-addressed and the content hash is returned; the caller is responsible for also recording that hash on record.

Returns:

The blob’s content hash if data was written, else None.

Raises:

RuntimeError – data was given but no blob store is configured.

exception lacing.BodySchemaError

Raised when a body fails validation against its registered schema.

class lacing.InMemoryOpLog

Simple list-backed op-log. Thread-safe via an RLock.

class lacing.IntervalAnnotationStore(*args, **kwargs)

Protocol for any interval-keyed annotation store.

Conceptually a MutableMapping[TimeInterval, list[Annotation]]: keys are TimeInterval; values are lists because multiple annotations can share an interval (different tiers, multiple annotators, soft labels).

We use Protocol rather than inheriting from MutableMapping so backends (in-memory, SQLite, Postgres) can structurally conform without forcing a single class hierarchy. The mapping methods below match the MutableMapping ABC; concrete backends like MemoryStore implement the full interface.

add(annotation: Annotation) → None

Append annotation to the list at its reference interval.

all() → Iterator[Annotation]

Iterate every annotation in the store, order unspecified.

at_tier(tier_name: str, query: TimeInterval) → Iterator[Annotation]

Annotations on tier_name that intersect query.

by_tier(tier_name: str) → Iterator[Annotation]

All annotations on tier_name, regardless of interval.

contains(query: TimeInterval) → Iterator[Annotation]

Annotations whose interval strictly contains query (Allen di).

during(query: TimeInterval) → Iterator[Annotation]

Annotations whose interval is strictly inside query (Allen d).

equals(query: TimeInterval) → Iterator[Annotation]

Allen =: identical interval.

extend(annotations: Iterable[Annotation]) → None

Add many; equivalent to repeated .add but adapters can optimize.

finishes(query: TimeInterval) → Iterator[Annotation]

Allen f: later start, same end.

intersects(query: TimeInterval) → Iterator[Annotation]

Annotations whose interval shares any time with query.

meets(query: TimeInterval) → Iterator[Annotation]

Allen m: a.end == q.start.

overlaps(query: TimeInterval) → Iterator[Annotation]

Strict Allen o: a.start < q.start < a.end < q.end.

relate(query: TimeInterval, relations: Iterable[AllenRelation]) → Iterator[Annotation]

Annotations whose interval has any of the named relations to query.

Generic dispatch — useful when relations are computed at runtime.

remove(annotation_id) → Annotation | None

Remove and return the annotation with this id, or None if absent.

starts(query: TimeInterval) → Iterator[Annotation]

Allen s: same start, earlier end.

tiers() → Iterator[Tier]

Registered tiers. Annotations may reference tiers not yet registered; callers decide whether that’s an error.

exception lacing.LossyTimeConversionError

Raised when a rate or seconds conversion would lose precision.

class lacing.MediaRef(*, kind: Literal['media'] = 'media', asset_id: str, interval: TimeInterval)

Reference to a region of a content-addressed media asset.

model_config = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class lacing.MemoryStore

IntervalAnnotationStore implementation over intervaltree.

Conforms to the protocol in lacing.store.base. We don’t formally inherit from IntervalAnnotationStore because it’s a Protocol with method bodies — structural typing is enough.

exception lacing.MigrationError

Raised when a migration step is missing or fails.

class lacing.NodeRef(*, kind: Literal['node'] = 'node', scene_path: str, interval: TimeInterval)

Reference to a node in a structured scene/document graph.

model_config = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class lacing.OpLog(*args, **kwargs)

Append-only log of mutations.

Implementations must guarantee that the clock returned by append() is strictly greater than every previously-returned clock value across the lifetime of the log.

append(operation: str, *, target_id: str | None = None, payload: dict[str, Any] | None = None, actor: str = 'anonymous') → int

Append an entry; return its assigned clock.

entries(*, until_clock: int | None = None, from_clock: int | None = None) → Iterator[OpLogEntry]

Iterate entries, optionally bounded by clock range (inclusive).

latest_clock() → int

Highest clock currently in the log; 0 if empty.

class lacing.OpLogEntry(clock: int, operation: str, target_id: str | None, payload: dict[str, ~typing.Any], actor: str = 'anonymous', received_at: float = <factory>)

One row of the op-log.

actor: str

user:<handle> or agent:<model>@<hash> or adapter:<format>.

clock: int

Monotonic Lamport clock starting at 1. Strictly increasing per log.

operation: str

add_annotation, remove_annotation, update_annotation, add_tier, set_meta, import_batch.

Type:

One of

payload: dict[str, Any]

JSON-serializable payload sufficient to replay the operation.

received_at: float

Wall-clock time the operation was received (seconds since epoch).

target_id: str | None

Annotation id, tier name, meta key, or None for batch ops.

exception lacing.ProcessorError

Raised when a registered processor’s invocation fails.

class lacing.Provenance(*, was_generated_by: str, was_attributed_to: str, was_derived_from: list[UUID] = <factory>, generated_at_time: RationalTime, activity: str = 'create')

W3C PROV-O subset, embedded inline on every annotation.

model_config = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class lacing.RationalTime(value: int, rate: int = 24000)

A point in time as value / rate seconds.

Immutable. Two RationalTime values with different rates compare via their rational value, so RationalTime(24, 24) == RationalTime(1, 1).

Examples

>>> RationalTime(24000) == RationalTime(1, 1)
True
>>> RationalTime.from_seconds("1.5", rate=2).value
3

classmethod from_seconds(seconds: float | Fraction | str, rate: int = 24000) → RationalTime

Build from seconds. Quantizes to rate; raises if lossy.

seconds may be a str like "1.001" to avoid float ingestion.

classmethod from_seconds_lossy(seconds: float | Fraction | str | int, *, rate: int = 24000, mode: str = 'round') → RationalTime

Build from seconds, quantizing to the nearest sample at rate.

Unlike from_seconds() — which raises LossyTimeConversionError when the value cannot be represented exactly — this method always succeeds by quantizing. mode selects the rounding rule:

• "round" — nearest sample, ties to even (default)

• "floor" — largest sample <= seconds

• "ceil" — smallest sample >= seconds

Use this when sample-level quantization is knowingly acceptable — the common case for user-supplied durations. Use from_seconds() when exactness matters and a lossy conversion should be an error.

Examples

>>> RationalTime.from_seconds_lossy("0.1", rate=3).value
0
>>> RationalTime.from_seconds_lossy("0.1", rate=3, mode="ceil").value
1

classmethod now(rate: int = 24000) → RationalTime

Wall-clock time as a RationalTime, quantized to rate.

Uses time.time_ns() and builds the value directly, sidestepping the float-quantization landmine of from_seconds(float). Every producer of an annotation or artifact needs this for Provenance.generated_at_time.

to_rate(new_rate: int) → RationalTime

Re-express at new_rate. Raises LossyTimeConversionError on loss.

to_seconds() → float

Float seconds — for display only. Never round-trip through this.

exception lacing.SchemaMismatchError

Raised when opening a .annot file with an incompatible schema.

class lacing.SqliteOpLog(path: str, *, check_same_thread: bool = True)

Op-log backed by a SQLite table.

Designed to share a database file with SqliteStore so the store snapshot + the op-log live together and survive a single cp project.annot project.backup step.

class lacing.SqliteStore(path: str | PathLike, *, check_same_thread: bool = True)

SQLite-backed IntervalAnnotationStore.

Parameters:

• path – Path to the .annot file. Use ":memory:" for an ephemeral in-memory database.

• check_same_thread – Forwarded to sqlite3.connect. We hold a single connection guarded by a lock; pass False when sharing across threads.

class lacing.Tier(name: str, *, stereotype: TierStereotype = TierStereotype.NONE, parent: str | None = None, metadata: dict | None = None)

A named annotation layer with optional parent and stereotype.

Tiers are pure metadata; they don’t own annotations. The store is keyed by interval, not by tier — annotations carry their tier name as a field. This matches ELAN’s TIME_ORDER indirection (see ANN-DOC §C).

class lacing.TierStereotype(value)

Constraints on how a child tier relates to its parent tier.

Names match ELAN exactly so EAF round-trips are trivial.

INCLUDED_IN = 'INCLUDED_IN'

Children lie within the parent but gaps between siblings are allowed.

NONE = 'NONE'

No parent constraint. Top-level tier.

SYMBOLIC_ASSOCIATION = 'SYMBOLIC_ASSOCIATION'

One-to-one association with parent; child shares parent’s interval exactly.

SYMBOLIC_SUBDIVISION = 'SYMBOLIC_SUBDIVISION'

Ordered subdivision; children share parent’s interval as a sequence (no times).

TIME_SUBDIVISION = 'TIME_SUBDIVISION'

Children fully partition the parent’s interval (no gaps, no overlap).

class lacing.TimeInterval(start: RationalTime, end: RationalTime)

A half-open interval [start, end).

start == end is a valid point annotation, not a degenerate case. Always start <= end; constructor raises ValueError otherwise.

property duration: RationalTime

end - start at the same rate as start.

exception lacing.UnknownBodySchemaError

Raised when an annotation’s body_schema_uri has no registered model.

lacing.boundary_iou(a: Iterable[TimeInterval], b: Iterable[TimeInterval]) → float

Mean IoU between two sets of intervals via greedy best-match.

For each interval in a, finds its highest-IoU match in b (without replacement — once a b interval is matched it’s removed from the pool). Unmatched intervals in either set contribute 0.0 to the mean.

Returns:

Mean IoU ∈ [0, 1]. Returns 0.0 if both sets are empty (defensible as a “no agreement to measure” baseline).

lacing.cohen_kappa(a: Sequence[T], b: Sequence[T]) → float

Cohen’s kappa for two annotators on a categorical label.

Parameters:

• a – Annotator A’s labels.

• b – Annotator B’s labels (must be the same length).

Returns:

κ ∈ [-1, 1]. 1 = perfect agreement, 0 = chance, negative = worse than chance.

Raises:

ValueError – If sequences differ in length or are empty.

Edge cases:

If only one category appears across both annotators, both observed and expected agreement are 1.0; we return 1.0 by convention.

lacing.export_json_schemas(target_dir: str | Path, *, overwrite: bool = True, include_meta: bool = True) → list[Path]

Write every registered schema as JSON files under target_dir.

Layout: <target_dir>/<name>/v<N>.json. Returns the list of paths written, in registration order.

Parameters:

• target_dir – Output directory. Created if missing.

• overwrite – If False, refuse to write a file that already exists.

• include_meta – If True, also write a <target_dir>/index.json mapping every URI to its file path and the Pydantic model’s qualified name (helps the codegen pipeline).

lacing.get_tracer(name: str = 'lacing', version: str | None = None) → Any

Return a tracer or a no-op fallback.

Parameters:

• name – Instrumentation name (typically __name__ of the caller’s module or a logical name like "lacing.server").

• version – Optional package version string.

Returns:

opentelemetry.trace.Tracer if OTel is installed, else a no-op object whose start_as_current_span() is a context manager yielding a no-op span.

lacing.hash_bytes(data: bytes) → str

Return the canonical asset_id (SHA-256 hex) for data.

>>> hash_bytes(b"hello world")[:8]
'b94d27b9'

lacing.hash_file(path: Path | str, *, chunk_size: int = 1048576) → str

Return the canonical asset_id (SHA-256 hex) for the file at path.

lacing.instrument_otel(app: Any, *, tracer_name: str = 'lacing.server') → Any

Add OpenTelemetry instrumentation to a FastAPI app.

Wraps every request in a span; tags the span with the response’s X-Lacing-Clock header value (when present) as lacing.clock.

No-op when OTel isn’t installed — the app is returned unchanged.

Parameters:

• app – The FastAPI app (from lacing.server.create_app()).

• tracer_name – Name passed to get_tracer().

Returns:

The same app, with middleware installed if OTel is available.

lacing.interval_iou(a: TimeInterval, b: TimeInterval) → float

Intersection-over-Union for two time intervals.

Returns 1.0 if both are equal point intervals at the same instant; 0.0 if they don’t intersect (including when only one is a point).

lacing.is_otel_active() → bool

Quick check: is OTel installed AND a TracerProvider configured?

lacing.json_schema(uri: str) → dict

Return the JSON Schema for the body model registered at uri.

Pydantic’s model_json_schema() output, unmodified.

lacing.krippendorff_alpha(annotations: ~collections.abc.Sequence[~collections.abc.Sequence[~lacing.quality.T | None]], *, distance: ~typing.Callable[[~lacing.quality.T, ~lacing.quality.T], float] = <function _nominal_distance>) → float

Krippendorff’s α across any number of annotators.

Parameters:

• annotations – A list of annotators, each a sequence of labels (one per unit). Use None for a missing annotation by that annotator on that unit.

• distance – Function (x, y) -> float measuring disagreement between two label values. Default is the nominal (0/1) distance.

Returns:

α. 1.0 = perfect agreement, 0.0 = chance.

Raises:

ValueError – If sequences differ in length, fewer than 2 annotators given, or fewer than 2 paired observations exist.

lacing.maybe_span(tracer: Any, name: str, **attributes: Any)

Open a span on tracer, attaching attributes if supported.

Works with both real OTel tracers and the no-op fallback.

lacing.migrate(body: dict, *, from_uri: str, to_uri: str) → dict

Migrate body from from_uri to to_uri via registered steps.

Composes single-step migrations. Raises MigrationError if any step is missing.

lacing.register_body_schema(uri: str, model: type[BaseModel]) → type[BaseModel]

Register model as the validator for uri. Returns model.

lacing.register_migration(*, schema_name: str, from_version: int, to_version: int)

Register a forward migration from v<from_version> to v<to_version>.

Decorated function takes a body dict and returns a new body dict. Migrations must be one major-version step at a time (to_version == from_version + 1).

Re-registering the same (schema_name, from_version) pair replaces the previous entry — convenient in tests, intentional for hot-reload.

lacing.register_processor(func: Callable[[...], Any] | None = None, *, name: str | None = None)

Register a processor under name (defaults to the function name).

The function may be sync or async; we wrap sync funcs to a coroutine.

lacing.registered_processors() → list[str]

Names of every registered processor, sorted.

lacing.render_artifact_exhibit(annotations: Iterable, *, out_dir: str | Path, formats: Sequence[str] = ('html', 'pdf', 'md'), title: str = 'Artifact exhibit', image_resolver: Callable[[Mapping], str | None] | None = None) → list[Path]

Render an annotation graph as a human-readable artifact exhibit.

Lays every annotation out as a card — body, panel images, and in-document hyperlinks to the artifacts it derives from / feeds into. HTML is authored; the PDF and Markdown derive from it.

Panel images are written once as content-addressed sibling files under <out_dir>/images/ and referenced relatively, so the HTML and Markdown stay small; the PDF embeds them and stays a single self-contained file. The images/ directory is created only when the graph actually has images.

Parameters:

• annotations – the lacing annotations to exhibit (any iterable). Order is preserved — pass them chain-ordered for a document that reads top-to-bottom.

• out_dir – directory the exhibit.{html,pdf,md} files land in.

• formats – which of html / pdf / md to write. The HTML is always built in memory (the others derive from it).

• title – document title.

• image_resolver – image-reference → local file path callback. Defaults to reading the reference’s own path field; a caller whose images live behind URLs passes a resolver that downloads / caches them (keeping this module media-agnostic).

Returns:

The written file paths.

Raises:

RuntimeError – when pdf / md is requested but its optional converter (weasyprint / dn) is not installed — the message names the install command.

lacing.replay_oplog(log: OpLog, *, until_clock: int | None = None, target_factory=None) → Any

Rebuild a store by replaying log up to (and including) until_clock.

Parameters:

• log – Source op-log.

• until_clock – Stop at this clock (inclusive). None = replay all.

• target_factory – Zero-arg callable returning a fresh empty store. Defaults to MemoryStore. Pass a SqliteStore factory to replay into a persistent file.

Returns:

The rebuilt store. Operations whose payload references unknown body schemas or tier parents are still applied; the caller is responsible for any post-replay validation.

async lacing.run_processor_async(name: str, *, store: Any, oplog: Any, **kwargs: Any) → Any

Run a processor in the current event loop. For async callers.

lacing.run_processor_sync(name: str, *, store: Any, oplog: Any, **kwargs: Any) → Any

Run a processor synchronously and return its result.

If the processor is async, we call it via asyncio.run (when no loop is running) or schedule and wait on it (when a loop is already active). Most callers from sync code want the former.

lacing.traced(tracer: Any, span_name: str | None = None, *, record_args: bool = False)

Decorator: wrap a function in a span on tracer.

Parameters:

• tracer – From get_tracer().

• span_name – Override the span name. Default: func.__qualname__.

• record_args – If True, attach (str-coerced) positional + keyword args as span attributes arg.<n> / kwarg.<name>. Off by default — args may contain large or sensitive data.

lacing.validate_body(body: dict, uri: str) → BaseModel

Validate body against the schema registered for uri.

Returns the parsed Pydantic instance. Raises BodySchemaError on validation failure (wrapping the underlying pydantic.ValidationError) or UnknownBodySchemaError if the URI isn’t registered.