Skip to content

schemacast.episode

Attributes

JsonScalar = str | int | float | bool | None module-attribute

KeyValue = str | int | float | bool module-attribute

MAX_KNOWN = 8 module-attribute

Classes

ContextItem

Bases: BaseModel

An earlier text in the same thread, shown for reference only.

Source code in schemacast/episode.py 
19
20
21
22
23
24
25
class ContextItem(BaseModel):
    """An earlier text in the same thread, shown for reference only."""

    model_config = ConfigDict(extra="forbid")

    source_id: str = Field(description="Identifier of the earlier text.")
    text: str = Field(description="Its text.")

Attributes

model_config = ConfigDict(extra='forbid') class-attribute instance-attribute
source_id = Field(description='Identifier of the earlier text.') class-attribute instance-attribute
text = Field(description='Its text.') class-attribute instance-attribute

Episode

Bases: BaseModel

One text to cast.

The structured half of a record — author, channel, thread — belongs in metadata and is written by an ordinary GraFlo resource, never read off the text by the model; metadata is not shown to the model.

Source code in schemacast/episode.py 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
class Episode(BaseModel):
    """One text to cast.

    The structured half of a record — author, channel, thread — belongs in
    ``metadata`` and is written by an ordinary GraFlo resource, never read off
    the text by the model; ``metadata`` is not shown to the model.
    """

    model_config = ConfigDict(extra="forbid")

    text: str = Field(min_length=1, description="The text to cast.")
    source_id: str = Field(
        min_length=1, description="Stable identifier of the source record."
    )
    reference_time: datetime | None = Field(
        default=None,
        description="When the text was written; resolves relative dates.",
    )
    metadata: dict[str, JsonScalar] = Field(
        default_factory=dict, description="Source metadata, never shown to the model."
    )
    context: list[ContextItem] = Field(
        default_factory=list,
        description="Preceding texts in the thread, for resolving references only.",
    )
    known: list[KnownEntity] = Field(
        default_factory=list,
        description=f"Entities the source names, at most {MAX_KNOWN}; an edge the "
        "text states may end at one.",
    )

    @field_validator("known")
    @classmethod
    def _distinct_known(cls, known: list[KnownEntity]) -> list[KnownEntity]:
        if len(known) > MAX_KNOWN:
            raise ValueError(f"at most {MAX_KNOWN} known entities, got {len(known)}")
        seen: set[tuple[str, tuple[tuple[str, str], ...]]] = set()
        for entity in known:
            identity = (
                entity.vertex_type,
                tuple(sorted((k, repr(v)) for k, v in entity.key.items())),
            )
            if identity in seen:
                raise ValueError(
                    f"{entity.vertex_type} {entity.key} is given more than once"
                )
            seen.add(identity)
        return known

Attributes

context = Field(default_factory=list, description='Preceding texts in the thread, for resolving references only.') class-attribute instance-attribute
known = Field(default_factory=list, description=f'Entities the source names, at most {MAX_KNOWN}; an edge the text states may end at one.') class-attribute instance-attribute
metadata = Field(default_factory=dict, description='Source metadata, never shown to the model.') class-attribute instance-attribute
model_config = ConfigDict(extra='forbid') class-attribute instance-attribute
reference_time = Field(default=None, description='When the text was written; resolves relative dates.') class-attribute instance-attribute
source_id = Field(min_length=1, description='Stable identifier of the source record.') class-attribute instance-attribute
text = Field(min_length=1, description='The text to cast.') class-attribute instance-attribute

EpisodeFileError

Bases: ValueError

A line of an episodes file that is not a valid episode.

Source code in schemacast/episode.py 
109
110
111
112
113
114
115
class EpisodeFileError(ValueError):
    """A line of an episodes file that is not a valid episode."""

    def __init__(self, path: Path, line: int, reason: str) -> None:
        super().__init__(f"{path}:{line}: {reason}")
        self.path = path
        self.line = line

Attributes

line = line instance-attribute
path = path instance-attribute

Methods:

__init__(path, line, reason)
Source code in schemacast/episode.py 
112
113
114
115
def __init__(self, path: Path, line: int, reason: str) -> None:
    super().__init__(f"{path}:{line}: {reason}")
    self.path = path
    self.line = line

KnownEntity

Bases: BaseModel

An entity the text belongs to or is about, identified by its source.

The model may relate to it by ref but never adds it as a vertex or states its properties: its identity comes from the source, never from the text.

Source code in schemacast/episode.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
class KnownEntity(BaseModel):
    """An entity the text belongs to or is about, identified by its source.

    The model may relate to it by ref but never adds it as a vertex or states
    its properties: its identity comes from the source, never from the text.
    """

    model_config = ConfigDict(extra="forbid")

    vertex_type: str = Field(min_length=1, description="Its vertex type.")
    key: dict[str, KeyValue] = Field(
        min_length=1,
        description="Its primary identity: every identity field of the type and "
        "nothing else, or it is not shown.",
    )
    role: str = Field(
        pattern=r"^[a-z][a-z_]*$",
        description="What it is to the text, e.g. ``subject``: the record the "
        "text belongs to.",
    )
    label: str | None = Field(default=None, description="A name shown to the model.")

    @field_validator("key")
    @classmethod
    def _no_empty_values(cls, key: dict[str, KeyValue]) -> dict[str, KeyValue]:
        empty = [name for name, value in key.items() if value is None or value == ""]
        if empty:
            raise ValueError(f"key fields without a value: {', '.join(empty)}")
        return key

Attributes

key = Field(min_length=1, description='Its primary identity: every identity field of the type and nothing else, or it is not shown.') class-attribute instance-attribute
label = Field(default=None, description='A name shown to the model.') class-attribute instance-attribute
model_config = ConfigDict(extra='forbid') class-attribute instance-attribute
role = Field(pattern='^[a-z][a-z_]*$', description='What it is to the text, e.g. ``subject``: the record the text belongs to.') class-attribute instance-attribute
vertex_type = Field(min_length=1, description='Its vertex type.') class-attribute instance-attribute

Functions:

read_episodes(path)

Read episodes from a JSON Lines file, one object per line.

Blank lines are skipped.

Raises:

Type Description
EpisodeFileError

on the first line that is not a valid episode.
Source code in schemacast/episode.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def read_episodes(path: str | Path) -> Iterator[Episode]:
    """Read episodes from a JSON Lines file, one object per line.

    Blank lines are skipped.

    Raises:
        EpisodeFileError: on the first line that is not a valid episode.
    """
    path = Path(path)
    with open(path, encoding="utf-8") as handle:
        for number, line in enumerate(handle, start=1):
            if not line.strip():
                continue
            try:
                yield Episode.model_validate(json.loads(line))
            except json.JSONDecodeError as error:
                raise EpisodeFileError(
                    path, number, f"not JSON: {error.msg}"
                ) from error
            except ValidationError as error:
                raise EpisodeFileError(path, number, str(error)) from error