airbytehq / airbytehq/PyAirbyte

Let sources define their own default document rendering processes

Ouverte
#74 0 commentaires 0 réactions 0 personnes assignées Voir sur GitHub
enhancement
Langage dominant
Python
Étoiles
344
Forks
77
Merge moyen
1 j 11 h
PR mergées (30 j)
35

Description

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,
)
```

Guide de contribution

Ouvrir le guide de contribution

Évaluation

Cette issue n'a pas encore été évaluée.

Recevez les nouvelles issues par e-mail

Un résumé court des issues GitHub adaptées aux débutants.