airbytehq / airbytehq/PyAirbyte

Let sources define their own default document rendering processes

未关闭
#74 0 条评论 0 个 reaction 已指派 0 人 在 GitHub 查看
enhancement
主要语言
Python
星标
344
派生
77
平均合并
1 天 11 小时
30 天内合并 PR
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 摘要。