airbytehq / airbytehq/PyAirbyte
Let sources define their own default document rendering processes
- 主要言語
- 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 はまだ評価されていません。