airbytehq / airbytehq/PyAirbyte

Let sources define their own default document rendering processes

オープン
#74 コメント 0 件 リアクション 0 件 担当者 0 名 GitHub で見る
enhancement
主要言語
Python
スター
344
フォーク
77
平均マージ
1日 11時間
マージ済み PR(30日)
35

説明

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

コントリビューションガイド

コントリビューションガイドを開く

評価

この issue はまだ評価されていません。

新しい issue をメールで受け取る

初心者向けの GitHub issue を短くまとめたダイジェスト。