airbytehq / airbytehq/PyAirbyte

Let sources define their own default document rendering processes

Offen
#74 0 Kommentare 0 Reaktionen 0 zugewiesene Personen Auf GitHub ansehen
enhancement
Vorherrschende Sprache
Python
Sterne
344
Forks
77
Ø Merge
1 T. 11 Std.
Gemergte PRs (30 T.)
35

Beschreibung

First mockup, was subtracted from initial launch of:

- https://github.com/airbytehq/PyAirbyte/pull/48

Here is a rough outline of how we imagine this would work:

```

To inform how to render a specific stream's records as documents, this implementation proposes that
sources define a `document_rendering` annotation in their JSON schema. This property would contain
instructions for how to render records as documents, such as which properties to render as content,
which properties to render as metadata, and which properties to render as annotations.

Assuming a stream like GitHub Issues, the `document_rendering` annotation might look like this:
```json
{
"airbyte_document_rendering": {
"title_property": "title",
"content_properties": ["body"],
"frontmatter_properties": ["url", "author"],
"metadata_properties": ["id", "created_at", "updated_at", "url"]
}
}
```

```
AIRBYTE_DOCUMENT_RENDERING = "airbyte_document_rendering"
TITLE_PROPERTY = "title_property"
CONTENT_PROPS = "content_properties"
METADATA_PROPERTIES = "metadata_properties"

class DocumentAutoRenderer(BaseModel):
"""Automatically render a stream's records as documents.

This class is a convenience class for automatically rendering a stream's records as documents.
It is a thin wrapper around the `DocumentRenderer` class, and is intended to be used when the
source does not provide a `document_rendering` annotation in its JSON schema."""

def __init__(self, stream_metadata: ConfiguredAirbyteStream) -> None:
"""Create a new DocumentAutoRenderer."""

render_instructions: dict | None = None
title_prop: str | None = None
content_props: list[str] = []
metadata_props: list[str] = []

if AIRBYTE_DOCUMENT_RENDERING in stream_metadata.stream.json_schema:
render_instructions = stream_metadata.stream.json_schema[AIRBYTE_DOCUMENT_RENDERING]
if TITLE_PROPERTY in render_instructions:
title_prop: str | None = render_instructions[TITLE_PROPERTY] or None
if CONTENT_PROPS in render_instructions:
content_props: list[str] = render_instructions[CONTENT_PROPS]
if METADATA_PROPERTIES in render_instructions:
metadata_props: list[str] = render_instructions[METADATA_PROPERTIES]

if stream_metadata.cursor_field:
cursor_prop: str | None = ".".join(stream_metadata.cursor_field)

super().__init__(
title_property=title_prop,
cursor_property=cursor_prop,
content_properties=[
key
for key, value in stream_metadata.json_schema.get("properties", {}).items()
if value.get("type") == "string"
],
metadata_properties=[
key
for key, value in stream_metadata.json_schema.get("properties", {}).items()
if value.get("type") != "string"
],
primary_key_properties=stream_metadata.primary_key,
)
```

Beitragsleitfaden

Beitragsleitfaden öffnen

Bewertung

Dieses Issue wurde noch nicht bewertet.

Neue Issues direkt in Ihr Postfach

Eine kurze Übersicht über anfängerfreundliche GitHub-Issues.