Skip to content

Commit a1db553

Browse files
feat: Added an extensible API for stream schema sources (meltano#3190)
* feat: Added an extensible API for stream schema sources * Read name from the instance instead * Update singer_sdk/schema/source.py Co-authored-by: sourcery-ai[bot] <58596630+sourcery-ai[bot]@users.noreply.github.com> * Document samples * Update tests/core/schema/test_source.py Co-authored-by: sourcery-ai[bot] <58596630+sourcery-ai[bot]@users.noreply.github.com> * Test accessing stream `schema` when it is not found * Catch errors to OpenAPI spec loading more holistically * Statement is now unreachable * Update discovery docs * Update packages and samples * refactor: Pass `*args` and `**kwargs` to superclass constructors * refactor: Mark `SchemaSource.get_schema` with `typing.final` * refactor: Ignore missing coverage for `Stream.schema_filepath` * ci: Check types of sample packages * docs: Remove redundant samples * chore: Fix some types * docs: Add migration guide for `schema_filepath` --------- Co-authored-by: sourcery-ai[bot] <58596630+sourcery-ai[bot]@users.noreply.github.com>
1 parent 35b499d commit a1db553

17 files changed

Lines changed: 865 additions & 94 deletions

File tree

docs/code_samples.md

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -159,8 +159,7 @@ class ParquetStream(Stream):
159159
name = parquet_schema.names[i]
160160
# Translate from the Parquet type to a JSON Schema type
161161
dtype = get_jsonschema_type(str(parquet_schema.types[i]))
162-
163-
# Add the new property to our list
162+
# Add the property to the list
164163
properties.append(th.Property(name, dtype))
165164

166165
# Return the list as a JSON Schema dictionary object

docs/guides/index.md

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ The following pages contain useful information for developers building on top of
77
88
porting
99
pagination-classes
10+
schema-sources
1011
custom-clis
1112
config-schema
1213
performance

docs/guides/schema-sources.md

Lines changed: 140 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,140 @@
1+
# Schema Sources
2+
3+
The Singer SDK provides an extensible API for loading schemas from various sources through the `SchemaSource` system. This enables loading schemas from:
4+
5+
- **File directories** - JSON schema files in a directory structure
6+
- **OpenAPI specifications** - Schema components from OpenAPI 3.1 specs (local or remote)
7+
- **Custom sources** - Implement your own schema loading logic
8+
9+
## Basic Usage
10+
11+
```python
12+
from typing import ClassVar
13+
14+
from singer_sdk import RESTStream, SchemaDirectory, StreamSchema
15+
16+
from myproject import schemas
17+
18+
# Create a schema source from a directory.
19+
# For example, if the package has the following directory structure:
20+
# myproject
21+
# └── schemas
22+
# ├── __init__.py
23+
# ├── projects.json
24+
# └── users.json
25+
# then the following code will load the schemas from the schemas subpackage:
26+
SCHEMAS_DIR = SchemaDirectory(schemas)
27+
28+
class ProjectsStream(RESTStream):
29+
name = "projects"
30+
schema: ClassVar[StreamSchema] = StreamSchema(SCHEMAS_DIR) # Loads from projects.json
31+
```
32+
33+
## OpenAPI Integration
34+
35+
```python
36+
from typing import ClassVar
37+
38+
from singer_sdk import OpenAPISchema, StreamSchema
39+
40+
# Load from OpenAPI spec
41+
openapi_source = OpenAPISchema("https://api.example.com/openapi.json")
42+
43+
class UsersStream(RESTStream):
44+
name = "users"
45+
schema: ClassVar[StreamSchema] = StreamSchema(openapi_source, key="User") # Load "User" component
46+
```
47+
48+
## Migrating from File-Path Based Schemas
49+
50+
If you're upgrading from an older version of the Singer SDK that used file paths for schema loading, here's how to migrate your streams to use the new schema sources system.
51+
52+
### Before: Using `schema_filepath`
53+
54+
```python
55+
from pathlib import Path
56+
from singer_sdk import RESTStream
57+
58+
SCHEMAS_DIR = Path(__file__).parent / "schemas"
59+
60+
class ProjectsStream(RESTStream):
61+
"""Projects stream with file-path based schema."""
62+
63+
name = "projects"
64+
schema_filepath = SCHEMAS_DIR / "projects.json" # Deprecated approach
65+
```
66+
67+
### After: Using `StreamSchema` with `SchemaDirectory`
68+
69+
```python
70+
from typing import ClassVar
71+
from singer_sdk import RESTStream, SchemaDirectory, StreamSchema
72+
from myproject import schemas # Your schemas module/package
73+
74+
SCHEMAS_DIR = SchemaDirectory(schemas)
75+
76+
class ProjectsStream(RESTStream):
77+
"""Projects stream with schema source."""
78+
79+
name = "projects"
80+
schema: ClassVar[StreamSchema] = StreamSchema(SCHEMAS_DIR) # New approach
81+
```
82+
83+
### Migration Steps
84+
85+
1. **Organize your schemas**: Ensure your schema files are in a Python package with an `__init__.py` file:
86+
87+
```
88+
myproject/
89+
├── __init__.py
90+
├── streams.py
91+
└── schemas/
92+
├── __init__.py
93+
├── projects.json
94+
└── users.json
95+
```
96+
97+
1. **Update imports**: Add the new schema source imports:
98+
99+
```python
100+
from singer_sdk import SchemaDirectory, StreamSchema
101+
from myproject import schemas
102+
```
103+
104+
1. **Create schema directory**: Replace file paths with a schema directory:
105+
106+
```python
107+
# Before
108+
SCHEMAS_DIR = Path(__file__).parent / "schemas"
109+
110+
# After
111+
SCHEMAS_DIR = SchemaDirectory(schemas)
112+
```
113+
114+
1. **Update stream classes**: Replace `schema_filepath` with the `StreamSchema` descriptor:
115+
116+
```python
117+
# Before
118+
class MyStream(RESTStream):
119+
schema_filepath = SCHEMAS_DIR / "my_stream.json"
120+
121+
# After
122+
class MyStream(RESTStream):
123+
schema: ClassVar[StreamSchema] = StreamSchema(SCHEMAS_DIR)
124+
```
125+
126+
1. **Handle custom schema keys** (if needed):
127+
128+
```python
129+
# If your stream name doesn't match the schema file name
130+
class ProjectDetailsStream(RESTStream):
131+
name = "project_details"
132+
schema: ClassVar[StreamSchema] = StreamSchema(SCHEMAS_DIR, key="ProjectDetail") # Uses ProjectDetail.json
133+
```
134+
135+
### Benefits of Migration
136+
137+
- **Better performance**: Schema caching reduces file I/O
138+
- **Type safety**: Improved type hints and validation
139+
- **Flexibility**: Support for multiple schema sources (files, OpenAPI, custom)
140+
- **Future-proof**: Access to new schema source features

docs/implementation/discovery.md

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -8,8 +8,9 @@ The catalog generated is automatically populated by a small number of developer
88
importantly:
99

1010
- `Tap.discover_streams()` - Should return a list of available "discovered" streams.
11-
- `Stream.schema` or `Stream.schema_filepath` - The JSON Schema definition of each stream,
12-
provided either directly as a Python `dict` or indirectly as a `.json` filepath.
11+
- `Stream.schema` - The JSON Schema definition of each stream,
12+
provided either directly as a Python `dict`, or through a
13+
[`StreamSchema`](/guides/schema-sources.md) descriptor with a schema source.
1314
- `Stream.primary_keys` - a list of strings indicating the primary key(s) of the stream.
1415
- `Stream.replication_key` - a single string indicating the name of the stream's replication
1516
key (if applicable).

packages/tap-countries/tap_countries/streams.py

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -9,11 +9,12 @@
99
from __future__ import annotations
1010

1111
import abc
12-
import importlib.resources
1312
import sys
13+
import typing as t
1414

1515
from requests_cache.session import CachedSession
1616

17+
from singer_sdk import SchemaDirectory, StreamSchema
1718
from singer_sdk import typing as th
1819
from singer_sdk.streams.graphql import GraphQLStream
1920
from tap_countries import schemas
@@ -24,7 +25,7 @@
2425
from typing import override # noqa: ICN003
2526

2627

27-
SCHEMAS_DIR = importlib.resources.files(schemas)
28+
SCHEMAS_DIR = SchemaDirectory(schemas)
2829

2930

3031
class CountriesAPIStream(GraphQLStream, metaclass=abc.ABCMeta):
@@ -102,7 +103,7 @@ class ContinentsStream(CountriesAPIStream):
102103

103104
name = "continents"
104105
primary_keys = ("code",)
105-
schema_filepath = SCHEMAS_DIR / "continents.json"
106+
schema: t.ClassVar[StreamSchema] = StreamSchema(SCHEMAS_DIR)
106107
query = """
107108
continents {
108109
code

packages/tap-dummyjson/tests/test_core.py

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,9 @@
11
from __future__ import annotations
22

3-
from samples.sample_tap_dummy_json.tap_dummyjson.tap import TapDummyJSON
43
from singer_sdk.testing import get_tap_test_class
54

5+
from tap_dummyjson.tap import TapDummyJSON
6+
67
CONFIG = {
78
"username": "emilys",
89
"password": "emilyspass",

packages/tap-gitlab/tap_gitlab/graphql_streams.py

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -6,9 +6,10 @@
66

77
from __future__ import annotations
88

9-
import importlib.resources
109
import sys
10+
import typing as t
1111

12+
from singer_sdk import SchemaDirectory, StreamSchema
1213
from singer_sdk.streams import GraphQLStream
1314
from tap_gitlab import schemas
1415

@@ -19,13 +20,14 @@
1920

2021
SITE_URL = "https://gitlab.com/graphql"
2122

22-
SCHEMAS_DIR = importlib.resources.files(schemas)
23+
SCHEMAS_DIR = SchemaDirectory(schemas)
2324

2425

2526
class GitlabGraphQLStream(GraphQLStream):
2627
"""Sample tap test for gitlab."""
2728

2829
url_base = SITE_URL
30+
schema: t.ClassVar[StreamSchema] = StreamSchema(SCHEMAS_DIR)
2931

3032
@property
3133
@override
@@ -44,7 +46,6 @@ class GraphQLCurrentUserStream(GitlabGraphQLStream):
4446
name = "currentuser"
4547
primary_keys = ("id",)
4648
replication_key = None
47-
schema_filepath = SCHEMAS_DIR / "currentuser.json"
4849
query = """
4950
currentUser {
5051
name
@@ -58,7 +59,6 @@ class GraphQLProjectsStream(GitlabGraphQLStream):
5859
name = "projects"
5960
primary_keys = ("id",)
6061
replication_key = None
61-
schema_filepath = SCHEMAS_DIR / "projects-graphql.json"
6262

6363
@property
6464
@override

packages/tap-gitlab/tap_gitlab/streams.py

Lines changed: 3 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -2,15 +2,14 @@
22

33
from __future__ import annotations
44

5-
import importlib.resources
65
import sys
76
import typing as t
87

98
from requests_cache import CachedSession
109

10+
from singer_sdk import RESTStream, SchemaDirectory, StreamSchema
1111
from singer_sdk.authenticators import SimpleAuthenticator
1212
from singer_sdk.pagination import SimpleHeaderPaginator
13-
from singer_sdk.streams.rest import RESTStream
1413
from singer_sdk.typing import (
1514
ArrayType,
1615
DateTimeType,
@@ -32,13 +31,14 @@
3231
from singer_sdk.helpers.types import Context
3332

3433

35-
SCHEMAS_DIR = importlib.resources.files(schemas)
34+
SCHEMAS_DIR = SchemaDirectory(schemas)
3635

3736

3837
class GitlabStream(RESTStream[str]):
3938
"""Sample tap test for gitlab."""
4039

4140
_LOG_REQUEST_METRIC_URLS = True
41+
schema: t.ClassVar[StreamSchema] = StreamSchema(SCHEMAS_DIR)
4242

4343
@property
4444
@override
@@ -129,7 +129,6 @@ class ProjectsStream(ProjectBasedStream):
129129
primary_keys = ("id",)
130130
replication_key = "last_activity_at"
131131
is_sorted = True
132-
schema_filepath = SCHEMAS_DIR / "projects.json"
133132

134133

135134
class ReleasesStream(ProjectBasedStream):
@@ -139,7 +138,6 @@ class ReleasesStream(ProjectBasedStream):
139138
path = "/projects/{project_id}/releases"
140139
primary_keys = ("project_id", "tag_name")
141140
replication_key = None
142-
schema_filepath = SCHEMAS_DIR / "releases.json"
143141

144142

145143
class IssuesStream(ProjectBasedStream):
@@ -150,7 +148,6 @@ class IssuesStream(ProjectBasedStream):
150148
primary_keys = ("id",)
151149
replication_key = "updated_at"
152150
is_sorted = False
153-
schema_filepath = SCHEMAS_DIR / "issues.json"
154151

155152

156153
class CommitsStream(ProjectBasedStream):
@@ -163,7 +160,6 @@ class CommitsStream(ProjectBasedStream):
163160
primary_keys = ("id",)
164161
replication_key = "created_at"
165162
is_sorted = False
166-
schema_filepath = SCHEMAS_DIR / "commits.json"
167163

168164

169165
class EpicsStream(ProjectBasedStream):
@@ -229,7 +225,6 @@ class EpicIssuesStream(GitlabStream):
229225
path = "/groups/{group_id}/epics/{epic_iid}/issues"
230226
primary_keys = ("id",)
231227
replication_key = None
232-
schema_filepath = SCHEMAS_DIR / "epic_issues.json"
233228
parent_stream_type = EpicsStream # Stream should wait for parents to complete.
234229

235230
@override

samples/aapl/aapl.py

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,8 +4,9 @@
44

55
import importlib.resources
66
import json
7+
import typing as t
78

8-
from singer_sdk import Stream, Tap
9+
from singer_sdk import SchemaDirectory, Stream, StreamSchema, Tap
910

1011
PROJECT_DIR = importlib.resources.files("samples.aapl")
1112

@@ -14,7 +15,10 @@ class AAPL(Stream):
1415
"""An AAPL stream."""
1516

1617
name = "aapl"
17-
schema_filepath = PROJECT_DIR / "fundamentals.json"
18+
schema: t.ClassVar[StreamSchema] = StreamSchema(
19+
SchemaDirectory(PROJECT_DIR),
20+
key="fundamentals",
21+
)
1822

1923
def get_records(self, _): # noqa: PLR6301
2024
"""Generate a single record."""

singer_sdk/__init__.py

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,12 @@
66
from singer_sdk.connectors import SQLConnector
77
from singer_sdk.mapper_base import InlineMapper
88
from singer_sdk.plugin_base import PluginBase
9+
from singer_sdk.schema.source import (
10+
OpenAPISchema,
11+
SchemaDirectory,
12+
SchemaSource,
13+
StreamSchema,
14+
)
915
from singer_sdk.sinks import BatchSink, RecordSink, Sink, SQLSink
1016
from singer_sdk.streams import GraphQLStream, RESTStream, SQLStream, Stream
1117
from singer_sdk.tap_base import SQLTap, Tap
@@ -15,6 +21,7 @@
1521
"BatchSink",
1622
"GraphQLStream",
1723
"InlineMapper",
24+
"OpenAPISchema",
1825
"PluginBase",
1926
"RESTStream",
2027
"RecordSink",
@@ -23,8 +30,11 @@
2330
"SQLStream",
2431
"SQLTap",
2532
"SQLTarget",
33+
"SchemaDirectory",
34+
"SchemaSource",
2635
"Sink",
2736
"Stream",
37+
"StreamSchema",
2838
"Tap",
2939
"Target",
3040
"streams",

0 commit comments

Comments
 (0)