Python API¶

def generate(
    document: dict[str, Any],
    *,
    prefix: str | None = None,
    include_validation: bool = False,
    documents: Mapping[str, dict[str, Any]] | None = None,
) -> tuple[str, list[str]]:
    """Return Python source and diagnostics. No generated code is executed.

    The schema title determines the extension class name and its canonical $id
    determines registration. Fields and applicable STAC objects are discovered
    from schema paths and references. Embedded offline validation is opt-in.

    Args:
        document: Full extension schema with a nonempty title and HTTP(S) $id.
        prefix: Optional namespace filter, without the colon.
        documents: Offline resources; missing external schemas are fetched over HTTP(S).
        include_validation: Embed the schema and a validate() method. Union
            decoding may require the schema even when this is false.

    Raises:
        SchemaError: If the schema, namespace, class name, or references cannot be supported.
    """
    document = bundle(document, documents=documents, fetch=fetch_http)
    _validate_document(json_graph(document))
    schema = Schema(document)
    fields = schema.discover(prefix)
    class_name = _class_name(document, fields, prefix)
    builder = _build_models(schema, fields, class_name)
    uri = document.get("$id")
    if isinstance(uri, str):
        uri = uri.removesuffix("#")
    if not isinstance(uri, str) or not uri.startswith(("https://", "http://")):
        raise SchemaError("The schema must declare an absolute HTTP(S) $id")
    # Python source must retain literal characters rather than HTML entities.
    env = Environment(  # nosec B701
        autoescape=False,
        loader=PackageLoader("pystac_ext_generator", "templates"),
        undefined=StrictUndefined,
        keep_trailing_newline=True,
        trim_blocks=True,
        lstrip_blocks=True,
    )
    env.filters["py"] = repr
    env.filters["optional"] = lambda value: _annotation_union(value, "None")
    env.filters["update_type"] = lambda value: _annotation_union(value, "_Unset", "None")
    env.filters["create_type"] = lambda value: _annotation_union(value, "_Unset")
    source = env.get_template("extension.py.j2").render(
        class_name=class_name,
        fields=fields,
        uri=uri,
        models=builder.models,
        include_validation=include_validation,
        codec_kinds=set().union(
            *(_codec_kinds(field.codec) for field in fields),
            *(_codec_kinds(member.codec) for model in builder.models for member in model.members),
        ),
        target_types=_target_types(fields),
        target_annotation=" | ".join("pystac." + name for name, _ in _target_types(fields)),
        document_json=json.dumps(json_graph(document), ensure_ascii=True, allow_nan=False),
        targets=sorted(set().union(*(f.targets for f in fields))),
        field_targets={f.key: sorted(f.targets) for f in fields},
        field_names={f.name: f.key for f in fields},
    )
    ast.parse(source)
    return _format_source(source), sorted(schema.warnings)

def bundle(
    document: dict[str, Any],
    *,
    documents: Mapping[str, dict[str, Any]] | None = None,
    fetch: Callable[[str], dict[str, Any]] | None = None,
) -> dict[str, Any]:
    """Return an eagerly dereferenced schema graph without mutating the input.

    Args:
        document: Root schema, whose $id provides the reference base URI.
        documents: Offline resources indexed by absolute URI.
        fetch: Loader for resources absent from documents; None disables fetching.

    Raises:
        SchemaError: If a reference cannot be resolved or a resource is not an object.
    """
    if not isinstance(document, dict):
        raise SchemaError("The schema must be a JSON object")
    resources = documents or {}
    dialect = document.get("$schema", "https://json-schema.org/draft/2020-12/schema")

    def load_resource(uri: str) -> dict[str, Any]:
        if uri in resources:
            return _prepare_document(resources[uri], dialect)
        if fetch is None:
            raise SchemaError(f"Missing external schema {uri}; supply --reference URI=FILE")
        return _prepare_document(fetch(uri), dialect)

    try:
        resolved = jsonref.replace_refs(
            _prepare_document(document, dialect),
            base_uri=document.get("$id") or "",
            loader=load_resource,
            jsonschema=True,
            proxies=False,
            lazy_load=False,
        )
    except (jsonref.JsonRefError, RecursionError) as exc:
        raise SchemaError(f"Could not resolve schema references: {exc}") from exc
    if not isinstance(resolved, dict):
        raise SchemaError("The schema must be a JSON object")
    _restore_schema(resolved, set(), root=True)
    return resolved

def fetch_http(uri: str) -> dict[str, Any]:
    """Fetch a JSON schema from an HTTP(S) URL with a 30-second timeout.

    Raises:
        SchemaError: If the URI is not HTTP(S) or the response is not an object.
    """
    if not uri.startswith(("https://", "http://")):
        raise SchemaError(f"Only HTTP(S) external references can be fetched: {uri}")
    # The scheme is checked above; urllib rejects file/custom-scheme redirects.
    with urlopen(uri, timeout=30) as response:  # nosec B310
        document = json.load(response)
    if not isinstance(document, dict):
        raise SchemaError(f"External schema must be an object: {uri}")
    return document