API Reference¶
Complete API documentation for sunstone-py.
Project Path Configuration¶
read_csv, read_excel, read_json, read_dataset, and the DataFrame constructor look up paths against a project directory containing datasets.yaml. The project_path argument is optional — when omitted, these functions fall back to the process-wide configured value, then to Path.cwd().
The configured value is stored in a contextvars.ContextVar, so it is safe across threads and async tasks.
sunstone.set_project_path(path)¶
Set the default project path for the current context.
Parameters:
path(str | Path): Project directory.
Returns: Path — the resolved absolute path.
Example:
sunstone.get_project_path()¶
Return the configured default project path, or Path.cwd() if none is set.
Returns: Path
sunstone.clear_project_path()¶
Clear any previously configured default project path.
Returns: None
sunstone.use_project_path(path)¶
Context manager that temporarily sets the default project path.
Parameters:
path(str | Path): Project directory.
Yields: Path — the resolved absolute path.
Example:
with sunstone.use_project_path('/path/to/other/project'):
df = pd.read_csv('inputs/data.csv')
# previous default restored here
Asset Envelope¶
Asset is the uniform container for every kind of data sunstone handles
— tabular, raster, n-D array, tile pyramid. sunstone.DataFrame is a
thin facade over an Asset of kind=AssetKind.TABULAR; the two record
identical lineage.
sunstone.read(path, *, format=None, kind=None, metadata=None, extras=None, **kwargs)¶
Read any registered format into an Asset.
Parameters:
path(str): Path or URL to read.format(str | None): Format override ('csv','parquet','zarr','hdf5','pdf','docx', a canonical MIME like'application/pdf', …). Auto-detected if not provided.kind(AssetKind | None): When supplied, overrides the kind on the returnedAsset. Use this when reconstructing an asset from a catalog row where the catalog is authoritative.metadata(Metadata | None): When supplied, fully replaces the handler-producedmetadata. Use this for catalog-driven reconstruction where the canonical metadata lives outside the file.extras(dict | None): When supplied, fully replaces the handler-producedextrasdict.**kwargs: Handler-specific keyword arguments.
Returns: Asset
Dispatch order:
- Explicit
kind=/format=arguments. - The
format:field on the matchingdatasets.yamlentry (resolved by location). - The path itself — directory paths route to the
StoreFormatHandlerregistry; single-file paths consult store handlers first (so HDF5 / NetCDF-4 can claim them), then fall through to the tabular stream pipeline.
Example:
import sunstone as ss
from sunstone.lineage import Metadata
asset = ss.read('inputs/era5_2024.zarr')
assert asset.kind is ss.AssetKind.ARRAY
arrays = asset.as_array() # dict[str, numpy.ndarray]
# Catalog-driven reconstruction: override handler-produced fields.
pdf_asset = ss.read(
'inputs/report.pdf',
metadata=Metadata(slug='q4-report', name='Q4 Report'),
extras={'media_type': 'application/pdf', 'origin': 'catalog'},
)
sunstone.write(asset, path, *, format=None, **kwargs)¶
Write an Asset to path. Dispatches via the plugin registry.
Parameters:
asset(Asset): Asset to write.path(str): Destination path or URL.format(str | None): Format override.**kwargs: Handler-specific keyword arguments.
Returns: None
Raises:
IncompatibleAssetKindError: If the selected handler does not supportasset.kind.ValueError: If no handler matchespathandformat.
AssetKind¶
Closed enum of supported asset kinds.
Values:
AssetKind.TABULAR—pandas.DataFrameAssetKind.RASTER—numpy.ndarray(single payload, e.g. GeoTIFF)AssetKind.ARRAY—dict[str, numpy.ndarray](multi-variable n-D arrays)AssetKind.TILES— tile pyramid descriptorAssetKind.BLOB—bytes(opaque binaries: PDF, DOCX, PPTX, RTF, plain text, etc.)
New kinds require adding a variant; plugin authors cannot extend the enum.
Asset¶
Uniform envelope across kinds.
Attributes:
payload(Any): Kind-native data.kind(AssetKind): Which kind this asset is.metadata(Metadata): Unified metadata container.extras(dict[str, Any]): Kind-specific accessory info (CRS, rasterio profile, chunk spec). Never copies of the payload.
Read-only convenience properties:
asset.profile→extras.get("profile")asset.crs→extras.get("crs")
Typed kind accessors (raise IncompatibleAssetKindError on mismatch):
asset.as_table() -> pandas.DataFrameasset.as_raster() -> numpy.ndarrayasset.as_array() -> dict[str, numpy.ndarray]asset.as_tiles() -> Anyasset.as_blob() -> bytes
Asset.derive(payload, *, slug=None, name=None, kind=None, derived_from=None, metadata_updates=None, extras_updates=None, inherit_custom_properties=False)¶
Return a new Asset derived from this one. Records
prov:wasDerivedFrom for each parent.
Parameters:
payload(Any): New payload for the child asset.slug/name(str | None): Identity for the child. Slug and name do not inherit from the parent.kind(AssetKind | None): Override the child's kind (default: same as parent).derived_from(Iterable[Asset] | None): Multi-parent provenance. Defaults to[self](single-parent derivation).metadata_updates(dict | None): ExtraMetadatakeys to set on the child.extras_updates(dict | None): Updates merged into the deep-copied parent extras.inherit_custom_properties(bool): IfTrue, copyparent.metadata.custom_propertiesto the child.
Returns: Asset
Example (multi-parent):
ndvi = (nir.as_raster() - red.as_raster()) / (nir.as_raster() + red.as_raster())
child = nir.derive(
ndvi,
slug='ndvi-2024',
name='NDVI 2024',
derived_from=[nir, red],
)
ss.write(child, 'outputs/ndvi.tif')
Per-kind derive policies (e.g. clearing stale rasterio profile
fields when shape or dtype changes) run after the metadata fork.
ComponentSchema¶
Per-component metadata: a tabular column, a raster band, an array variable, or a tile layer.
Attributes:
name(str): Component name.component_kind(str): One of"column","band","variable","layer".dtype(str | None): Data type, e.g."float32","uint16".units(str | None): Pint-parsable unit string, e.g."kelvin","m/s".description(str | None): Human-readable description.
Stored on Metadata.component_metadata keyed by component name.
IncompatibleAssetKindError¶
Raised when a typed accessor (as_table, as_raster, ...) or a handler
is called with an asset whose kind does not match.
pandas Module¶
Drop-in replacement for pandas with lineage tracking.
Functions¶
read_dataset(slug, project_path=None, strict=None, fetch_from_url=True, format=None, **kwargs)¶
Read a dataset by slug with automatic format detection.
Parameters:
slug(str): Dataset slug to look up indatasets.yamlproject_path(str | Path | None): Path to project directory. Defaults to the value set withsunstone.set_project_path(), orPath.cwd()if none is setstrict(bool | None): Enable strict mode. If None, reads fromSUNSTONE_DATAFRAME_STRICTenv varfetch_from_url(bool): If True and dataset has a source URL but no local file, fetch automaticallyformat(str | None): Format override ('csv','json','excel','parquet','tsv'). Auto-detected from extension if not provided**kwargs: Additional arguments passed to the underlying pandas reader
Returns: DataFrame with lineage tracking
Example:
df = pd.read_dataset('official-un-member-states')
df = pd.read_dataset('my-data', format='json', project_path='/path/to/project')
read_csv(filepath, project_path=None, strict=None, **kwargs)¶
Read CSV file with lineage tracking.
Parameters:
filepath(str | Path): Path to CSV file, URL, or dataset slugproject_path(str | Path | None): Path to project directory containingdatasets.yaml. Defaults to the value set withsunstone.set_project_path(), orPath.cwd()if none is setstrict(bool | None): If True, dataset must be pre-registered indatasets.yaml. If None, reads fromSUNSTONE_DATAFRAME_STRICTenv var**kwargs: Additional arguments passed topandas.read_csv()
Returns: DataFrame with lineage tracking
Raises:
DatasetNotFoundError: If dataset not found indatasets.yamlStrictModeError: If strict=True and dataset not pre-registered
Example:
df = pd.read_csv(
'data/schools.csv',
project_path='/path/to/project',
strict=True,
encoding='utf-8'
)
read_excel(filepath, project_path=None, strict=None, fetch_from_url=True, **kwargs)¶
Read Excel file (.xlsx/.xls) with lineage tracking.
Parameters:
filepath(str | Path): Path to Excel file or dataset slugproject_path(str | Path | None): Path to project directory containingdatasets.yaml. Defaults to the value set withsunstone.set_project_path(), orPath.cwd()if none is setstrict(bool | None): If True, dataset must be pre-registered. If None, reads fromSUNSTONE_DATAFRAME_STRICTenv varfetch_from_url(bool): If True and dataset has a source URL but no local file, automatically fetch from URL**kwargs: Additional arguments passed topandas.read_excel()
Returns: DataFrame with lineage tracking
Raises:
DatasetNotFoundError: If dataset not found indatasets.yamlFileNotFoundError: Ifdatasets.yamldoesn't exist
Example:
# Load by slug (recommended)
df = pd.read_excel('my-excel-data', project_path='/path/to/project')
# Load by file path
df = pd.read_excel('data/schools.xlsx', project_path='/path/to/project', sheet_name='Sheet1')
read_json(filepath, project_path=None, strict=None, **kwargs)¶
Read JSON file with lineage tracking.
Parameters:
filepath(str | Path): Path to JSON file or dataset slugproject_path(str | Path | None): Path to project directory. Defaults to the value set withsunstone.set_project_path(), orPath.cwd()if none is setstrict(bool | None): Enable strict mode. If None, reads fromSUNSTONE_DATAFRAME_STRICTenv var**kwargs: Additional arguments passed topandas.read_json()
Returns: DataFrame with lineage tracking
Example:
# Read a JSON file
df = pd.read_json('data/records.json', project_path=PROJECT_PATH)
# With pandas options
df = pd.read_json('data/records.json', orient='records', lines=True)
merge(left, right, **kwargs)¶
Merge DataFrames with combined lineage.
Parameters:
left(DataFrame): Left DataFrameright(DataFrame): Right DataFrame**kwargs: Arguments passed topandas.merge()
Returns: DataFrame with lineage from both sources
Example:
result = pd.merge(schools, teachers, on='school_id', how='inner')
print(len(result.lineage.sources)) # 2
concat(dfs, **kwargs)¶
Concatenate DataFrames with combined lineage.
Parameters:
dfs(list[DataFrame]): List of DataFrames to concatenate**kwargs: Arguments passed topandas.concat()
Returns: DataFrame with lineage from all sources
Example:
DataFrame Class¶
Main class for working with data and lineage.
Class Methods¶
read_csv(filepath, project_path, strict=False, **kwargs)¶
Read CSV file and return DataFrame.
Parameters: Same as pandas.read_csv()
Returns: DataFrame instance
read_excel(filepath, project_path, strict=False, fetch_from_url=True, **kwargs)¶
Read Excel file and return DataFrame.
Parameters: Same as pandas.read_excel()
Returns: DataFrame instance
Instance Methods¶
to_csv(path, slug, name, *, license=None, check_license=True, track=True, **kwargs)¶
Write DataFrame to CSV and register in datasets.yaml.
Parameters:
path(str | Path): Output file pathslug(str | None): Machine-readable identifier (required in relaxed mode if not registered)name(str | None): Human-readable name (required in relaxed mode if not registered)license(str | None): SPDX license identifier for the output. Persisted todatasets.yamland used as the target for the compatibility check. When omitted, falls back to the dataset's existinglicense, then topackages[].license/package.license. If still unresolved and source licenses exist, one is auto-derived (inherited from a single source, or the most restrictive license that satisfies all sources) and persisted todatasets.yaml.check_license(bool): IfTrue(default), raiseLicenseCompatibilityErrorwhen the effective target license is incompatible with any source license collected from the current session lineage. Passcheck_license=Falseto skip the check (and the auto-derivation).track(bool): IfFalse, write the CSV directly without lineage tracking, dataset registration, or license enforcement. Useful for tests and exploratory work.**kwargs: Arguments passed topandas.DataFrame.to_csv()
Returns: None
Raises:
StrictModeError: In strict mode, if dataset not registered.ValueError: In relaxed mode, ifslug/namenot provided for a new dataset.LicenseCompatibilityError: Ifcheck_licenseisTrueand either (a) the declared target license conflicts with a source license, or (b) no target was declared and the source licenses are mutually incompatible / unverifiable so no default can be derived.
Example:
df.to_csv(
'outputs/summary.csv',
slug='summary',
name='Summary Results',
license='CC-BY-4.0',
index=False,
)
Note: Publishing is controlled by the top-level publish configuration in datasets.yaml, not per-dataset.
to_parquet(path, slug, name, *, license=None, check_license=True, track=True, **kwargs)¶
Write DataFrame to Parquet file and register in datasets.yaml.
Parameters:
path(str | Path): Output file pathslug(str | None): Machine-readable identifier (required in relaxed mode if not registered)name(str | None): Human-readable name (required in relaxed mode if not registered)license(str | None): SPDX license identifier for the output. Same semantics asto_csv'slicenseparameter.check_license(bool): IfTrue(default), enforce license compatibility against source licenses. Seeto_csv.track(bool): IfFalse, write without lineage tracking, dataset registration, or license enforcement.**kwargs: Arguments passed topandas.DataFrame.to_parquet()
Returns: None
Raises: Same as to_csv.
Example:
df.to_parquet(
'outputs/summary.parquet',
slug='summary',
name='Summary Results',
license='CC-BY-4.0',
)
set_field_metadata(column, *, description, unit, source, type, constraints)¶
Set metadata for a column. Returns self for method chaining.
Parameters:
column(str): Column name to annotatedescription(str, optional): Human-readable description of the fieldunit(str, optional): Unit of measure (e.g.,'kg','students','%')source(str, optional): Slug of the input dataset this field comes fromtype(str, optional): Data type override. If None, inferred from dtype at write timeconstraints(dict, optional): Validation constraints (e.g., enum values)
Returns: DataFrame (self, for chaining)
Example:
df.set_field_metadata('population', description='Total population', unit='people')
df.set_field_metadata('gdp', description='Gross domestic product', unit='USD')
# Method chaining
df = (df
.set_field_metadata('area', unit='km^2')
.set_field_metadata('density', unit='people / km^2')
)
merge(right, **kwargs)¶
Merge with another DataFrame.
Parameters:
right(DataFrame): DataFrame to merge with**kwargs: Arguments passed topandas.merge()
Returns: New DataFrame with combined lineage
join(other, **kwargs)¶
Join with another DataFrame.
Parameters:
other(DataFrame): DataFrame to join with**kwargs: Arguments passed topandas.DataFrame.join()
Returns: New DataFrame with combined lineage
concat(others, **kwargs)¶
Concatenate with other DataFrames.
Parameters:
others(list[DataFrame]): DataFrames to concatenate**kwargs: Arguments passed topandas.concat()
Returns: New DataFrame with combined lineage
apply_operation(operation, description)¶
Apply transformation with lineage tracking.
Parameters:
operation(callable): Function that takes a pandas DataFrame and returns a pandas DataFramedescription(str): Human-readable description of the operation
Returns: New DataFrame with operation recorded in lineage
Example:
def adjust_enrollment(df):
return df.assign(adjusted=df['enrollment'] * 1.1)
result = df.apply_operation(
adjust_enrollment,
description="Apply 10% enrollment adjustment factor"
)
Instance Attributes¶
data¶
Access the underlying pandas DataFrame.
Type: pandas.DataFrame
Example:
# Get numpy array
values = df.data.values
# Use pandas methods not wrapped
styled = df.data.style.highlight_max()
metadata¶
Access the unified metadata container.
Type: Metadata
Example:
# Lineage is accessed through metadata
print(df.metadata.lineage.sources)
print(df.metadata.lineage.get_licenses())
df.metadata.slug = 'my-dataset'
df.metadata.name = 'My Dataset'
df.metadata.description = 'A description of this dataset'
# RDF prefixes and custom properties
df.metadata.rdf_prefixes = {'schema': 'http://schema.org/'}
df.metadata.custom_properties = {'schema:about': 'Education'}
# Per-field metadata (see set_field_metadata)
print(df.metadata.field_metadata)
lineage (deprecated)¶
Access lineage metadata directly. Use df.metadata.lineage instead.
Type: LineageMetadata
Example:
DatasetsManager Class¶
Manage datasets.yaml files programmatically.
Constructor¶
DatasetsManager(project_path, datasets_file=None)¶
Create a datasets manager.
Parameters:
project_path(str | Path): Path to project directory containingdatasets.yamldatasets_file(str | Path | None): Path to a specific datasets YAML file (relative to project_path or absolute). Defaults to"datasets.yaml"
Example:
manager = DatasetsManager('/path/to/project')
# Use a custom datasets file
manager = DatasetsManager('/path/to/project', datasets_file='config/my-datasets.yaml')
Methods¶
find_dataset_by_location(location, dataset_type=None)¶
Find dataset by file path.
Parameters:
location(str): File path to search fordataset_type(str, optional): Filter by 'input' or 'output'
Returns: DatasetMetadata | None
Example:
find_dataset_by_slug(slug, dataset_type=None)¶
Find dataset by slug identifier.
Parameters:
slug(str): Slug to search fordataset_type(str, optional): Filter by 'input' or 'output'
Returns: DatasetMetadata | None
Example:
get_all_inputs()¶
Get all input datasets.
Returns: list[DatasetMetadata]
get_all_outputs()¶
Get all output datasets.
Returns: list[DatasetMetadata]
get_publish_config()¶
Get the top-level publish configuration.
Returns: PublishConfig | None
Example:
publish_config = manager.get_publish_config()
if publish_config and publish_config.enabled:
print(f"Publishing to: {publish_config.to}")
print(f"Flatten: {publish_config.flatten}")
add_output_dataset(name, slug, location, fields)¶
Register new output dataset.
Parameters:
name(str): Human-readable nameslug(str): Machine-readable identifierlocation(str): File pathfields(list[FieldSchema]): Field definitions
Returns: None
Example:
from sunstone import FieldSchema
manager.add_output_dataset(
name='Analysis Results',
slug='analysis-results',
location='outputs/results.csv',
fields=[
FieldSchema(name='category', type='string'),
FieldSchema(name='count', type='integer'),
FieldSchema(name='avg_value', type='number')
]
)
Note: Use the top-level publish configuration in datasets.yaml to enable publishing for all outputs.
update_output_dataset(slug, **kwargs)¶
Update existing output dataset.
Parameters:
slug(str): Dataset slug to update**kwargs: Fields to update (name, location, fields, etc.)
Returns: None
set_dataset_strict(slug, strict, dataset_type=None)¶
Enable or disable strict mode for a dataset.
Parameters:
slug(str): Dataset slugstrict(bool): True to enable strict mode, False to disabledataset_type(str, optional): Filter by 'input' or 'output'
Returns: None
Raises: DatasetNotFoundError if dataset not found
Example:
# Enable strict mode
manager.set_dataset_strict('school-data', True)
# Disable strict mode
manager.set_dataset_strict('school-data', False)
update_output_lineage(slug, lineage, content_hash, strict=False)¶
Update lineage metadata for an output dataset.
Parameters:
slug(str): Output dataset sluglineage(LineageMetadata): Lineage metadata to writecontent_hash(str): Hash of the file contentstrict(bool): If True, validates without modifying
Returns: None
Raises:
DatasetNotFoundError: If dataset not foundDatasetValidationError: In strict mode, if lineage differs
Note: Timestamp only updates when content_hash changes.
get_absolute_path(location)¶
Convert relative path to absolute project path.
Parameters:
location(str): Relative or absolute path
Returns: Path
Process-level cache¶
By default every to_csv/to_parquet/write call constructs a fresh DatasetsManager,
which re-parses datasets.yaml (plus includes and the lock file) each time. For pipelines
that write many outputs in one process, wrap the work in datasets_manager_cache() to share
a single manager per resolved datasets.yaml path.
from sunstone import datasets_manager_cache
with datasets_manager_cache():
for df in frames:
df.to_parquet(...) # reuses one cached manager
datasets_manager_cache()¶
Context manager that activates a per-resolved-path manager cache for its duration. Nesting is
safe (inner uses share the outer cache; the cache is cleared on the outermost exit). Cached
managers are mtime-checked, so an external edit to datasets.yaml is picked up automatically.
get_datasets_manager(project_path, datasets_file=None)¶
Cache-aware acquisition. With no active datasets_manager_cache() context it constructs a
fresh DatasetsManager (unchanged legacy behavior); inside a context it returns a shared
instance keyed by the resolved datasets file path.
clear_datasets_manager_cache()¶
Drop any cached managers. Mainly useful for test isolation.
Linting¶
lint_project(project_path, *, datasets_file='datasets.yaml', rules=None)¶
Lint a Sunstone project's datasets.yaml against the Sunstone Minimum Viable Metadata recommendations.
Parameters:
project_path(str | Path): Project directory or a direct path to adatasets.yamlfile.datasets_file(str): Filename relative toproject_path. Defaults to'datasets.yaml'.rules(set[str] | None): Optional set of rule IDs to run (e.g.{'R005', 'R104'}).Noneruns all rules.
Returns: LintReport — see below.
Example:
from sunstone import lint_project
report = lint_project('/path/to/project')
for v in report.errors:
print(f"{v.rule_id} {v.location}: {v.message}")
See the CLI Guide for the full rule list.
LintReport¶
Aggregated lint output.
Attributes:
project_path(Path): The linted project directory.violations(list[Violation]): Active findings (not suppressed bylint.disable).suppressed(list[Violation]): Findings silenced bylint.disable, kept around for audit.suppressions(dict[str, str]): Map ofrule_id→ justification string fromlint.disable.
Properties:
errors(list[Violation]): Subset ofviolationswithseverity == ERROR.warnings(list[Violation]): Subset ofviolationswithseverity == WARNING.info(list[Violation]): Subset ofviolationswithseverity == INFO.
Methods:
to_dict()— JSON-serialisable summary including counts and violation details.format_text()— Human-readable text summary.
Violation¶
A single rule violation.
Attributes:
rule_id(str): Stable identifier (e.g.'R005').severity(Severity):ERROR,WARNING, orINFO.message(str): Short description of the problem.location(str): Path withindatasets.yaml(e.g.'inputs[0].source.license').fix_hint(str | None): Suggestion for how to fix it.
Validation Functions¶
check_notebook_imports(notebook_path)¶
Validate a single notebook's imports.
Parameters:
notebook_path(str | Path): Path to notebook file
Returns: ValidationResult
Example:
result = check_notebook_imports('analysis.ipynb')
if result.is_valid:
print("✓ Notebook uses sunstone imports")
else:
print(result.summary())
validate_project_notebooks(project_path)¶
Validate all notebooks in a project.
Parameters:
project_path(str | Path): Path to project directory
Returns: dict[Path, ValidationResult]
Example:
results = validate_project_notebooks('/path/to/project')
for path, result in results.items():
if not result.is_valid:
print(f"\n{path}:")
print(result.summary())
Data Classes¶
FieldSchema¶
Field definition for datasets.
Attributes:
name(str): Field nametype(str | None): Field type (string, number, integer, boolean, date, datetime). If None, inferred from dtype at write timedescription(str, optional): Field descriptionunit(str, optional): Unit of measure (e.g.,'kg','%','people')source(str, optional): Slug of the input dataset this field's data comes fromconstraints(dict, optional): Validation constraints
Example:
from sunstone import FieldSchema
field = FieldSchema(
name='enrollment',
type='integer',
description='Number of enrolled students',
unit='students',
constraints={'minimum': 0}
)
# type can be omitted — it's inferred at write time
field = FieldSchema(name='ratio', description='Student-teacher ratio')
DatasetMetadata¶
Dataset metadata from datasets.yaml.
Attributes:
name(str): Human-readable nameslug(str): Machine-readable identifierlocation(str): File pathfields(list[FieldSchema]): Field definitionssource(SourceMetadata | None): Source attribution (inputs only)strict(bool): Strict mode enableddataset_type(str): 'input' or 'output'
PublishConfig¶
Top-level publishing configuration.
Attributes:
enabled(bool): Whether publishing is enabledto(str | None): Destination URL or pathflatten(bool): Whether to flatten directory structure (default: False)
Path Resolution:
- If
toends with.json: Used as datapackage filename gs://bucket/countries.json→ datapackage at exact path- If
todoesn't end with.json: Treated as directory gs://bucket/datasets/project/→ adds/datapackage.json
Example:
from sunstone import PublishConfig
config = PublishConfig(
enabled=True,
to='gs://my-bucket/datasets/project/',
flatten=False
)
LineageMetadata¶
Lineage tracking information. Aligned with W3C PROV-O.
Attributes:
sources(list[DatasetMetadata]): Source datasets that contributed to this datacreated_at(datetime | None): Timestamp when lineage was last updated (content changed)content_hash(str | None): SHA256 hash of the DataFrame contentactivity(Activity | None): The PROV-O Activity that generated this datafield_derivations(list[FieldDerivation] | None): Field-level derivation detail (prov:qualifiedDerivation)
Methods:
get_licenses(): Return list of all source licensesadd_source(source): Add source datasetpopulate_field_derivations(columns, slug): Auto-populate field derivations for columns from a sourcemerge(other): Merge lineage from another DataFrame, combining sources and field derivations
Activity¶
A W3C PROV-O Activity representing a script or notebook execution.
Attributes:
id(str): Unique identifier (e.g.,'exec-{timestamp}-{hash}')used(list[UsageRecord]): Input entities consumed by this activitygenerated(list[EntityRef]): Output entities producedwas_associated_with(list[Agent]): Agents involved in this activitystarted_at(datetime | None): When the activity startedended_at(datetime | None): When the activity endedscript_path(str | None): Path to the executed Python scriptgit_commit(str | None): Git commit hash at time of execution
Agent¶
A W3C PROV-O Agent: something that bears responsibility for an activity.
Attributes:
id(str): Unique identifier (username, org name, software name)type(AgentType): One ofPERSON,SOFTWARE,ORGANIZATIONlabel(str | None): Human-readable labelversion(str | None): Version string (for SoftwareAgent)
FieldDerivation¶
Records that an output field was derived from a source entity. Maps to prov:qualifiedDerivation at the field level.
Attributes:
output_field(str): Name of the output columnsource_entity(str): Slug of the source datasetsource_field(str | None): Name of the source field, if known
EntityRef¶
Lightweight reference to a PROV Entity (dataset).
Attributes:
slug(str): Dataset slug identifiernamespace(str | None): Optional namespace URI for external entities
UsageRecord¶
Records how an Activity used an Entity. Maps to prov:qualifiedUsage.
Attributes:
entity(EntityRef): Which entity was usedcolumns(list[str] | None): Which columns were selected (None means all)filters(dict | None): Filters applied during read
Metadata Class¶
Unified metadata container for DataFrames.
Attributes:
lineage(LineageMetadata): Lineage metadata tracking data provenancedescription(str | None): Human-readable description of the datasetrdf_prefixes(dict | None): RDF namespace prefixes for custom propertiescustom_properties(dict | None): Custom properties including RDF triplesfield_metadata(dict[str, FieldSchema]): Per-column metadata, keyed by column name (tabular shortcut overcomponent_metadata)component_metadata(dict[str, ComponentSchema]): Per-component metadata keyed by component name (column, band, variable, layer)slug(str | None): Dataset slug, used at write timename(str | None): Human-readable dataset name, used at write time
Plugin System¶
The plugin system handles reading, writing, and URL resolution through a registry of handlers.
PluginRegistry¶
Central registry for auth providers, URL handlers, and format handlers.
PluginRegistry.get(project_path=None)¶
Return a cached registry instance. If project_path is provided, the registry is scoped to that project and loads project-specific plugin configuration.
Example:
registry.fetch(url, dest)¶
Download a URL to a local file using the appropriate URL handler.
Parameters:
url(str): URL to download (supportshttp://,https://,gs://,s3://,r2://, local paths)dest(Path): Local destination file path
Returns: Path to the downloaded file
Example:
from pathlib import Path
from sunstone.plugins import PluginRegistry
registry = PluginRegistry.get()
registry.fetch('gs://my-bucket/data.csv', Path('data/local.csv'))
Note:
DatasetsManager.fetch_from_url()is deprecated. UsePluginRegistry.get().fetch()instead.
Content-type discovery¶
Four accessors expose what sunstone knows how to read or write so downstream tools (catalogs, dispatch layers, UIs) can enumerate the format surface without re-implementing the format catalogue.
registry.known_content_descriptors() -> set[ContentDescriptor]¶
Union of content_descriptors() across every registered format and
store handler that declares the method. Handlers without the method
contribute nothing.
registry.known_content_types() -> set[str]¶
Convenience projection — the set of canonical MIME strings present in
known_content_descriptors(), regardless of encoding.
registry.known_extensions() -> dict[str, FormatHandler | StoreFormatHandler]¶
Map of declared file extension → handler. First-registered wins on conflict, matching dispatch priority (external plugins win over built-ins on overlapping extensions).
registry.handler_for_content(content_type, content_encoding=None)¶
First handler whose declared content_descriptors() contains a
matching (content_type, content_encoding) pair.
- Parameters are stripped from
content_type(so"text/csv; charset=utf-8"matches"text/csv"). content_encoding=Nonematches identity-encoded handlers; passing"gzip","br", etc. requires a handler that declares the encoded variant explicitly.- Returns
Nonewhen nothing matches.
from sunstone.plugins import PluginRegistry
reg = PluginRegistry.get()
reg.known_content_types() # {'text/csv', 'application/pdf', ...}
reg.handler_for_content('text/csv; charset=utf-8') # <BuiltinFormatHandler ...>
reg.handler_for_content('text/csv', 'gzip') # None
ContentDescriptor¶
Frozen dataclass identifying what a handler reads or writes. Mirrors
HTTP Content-Type + Content-Encoding.
Attributes:
content_type(str): Canonical MIME, no parameters (e.g."text/csv").content_encoding(str | None): Transport/storage encoding —"gzip","br","zstd","deflate", orNonefor identity.
ContentDescriptor is hashable and equal-by-value, so the
known_content_descriptors() view is a real set.
Plugin Protocols¶
Plugins implement one or more of these protocols:
AuthProvider: Provides authentication headers for HTTP requests.URLHandler: Resolves URLs to readable/writable streams viaopen(url, mode).FormatHandler: Stream-based format reader/writer. Used for single-file formats whose library accepts a byte stream (CSV, JSON, Parquet,.npz, PDF/DOCX/etc. viaBlobFormatHandler). Returns / acceptsAsset.ContentDescriptorAware(optional,sunstone.plugins): Extends anyFormatHandler/StoreFormatHandlerwithcontent_descriptors()andextensions()methods so it participates in content-type discovery. Handlers without these methods still dispatch normally; they just don't appear in enumeration.StoreFormatHandler(sunstone.resource.StoreFormatHandler): Path-based format reader/writer for formats whose library needs a real path or directory (HDF5, Zarr, MBTiles, partitioned Parquet). Handlers MUST declare__sunstone_handler_protocol__ = 2and operate on aResourceLocation. Single-file store handlers are consulted before the stream registry, so they can claim paths likedata.h5.CLIProvider: Contributes atypercommand group mounted undersunstone <name>. Built-in groups (dataset,package,lineage,env,license) are protected from collision.
ResourceLocation¶
Wraps a path or URL that may refer to a single file or a
directory/prefix. Pass to StoreFormatHandler.read() / .write().
Attributes:
path(str): The underlying path or URL string.
Methods:
as_path() -> pathlib.Path— local-path view.is_dir() -> bool— true when the path resolves to a directory.list(glob='*') -> Iterator[ResourceLocation]— enumerate children.subpath(rel) -> ResourceLocation— join a relative path.open_byte_stream(mode='rb') -> BinaryIO— single-file byte stream (routes through theURLHandlerregistry for non-local paths).
Plugin Discovery¶
External plugins are discovered via the sunstone.plugins entry point group:
# In your plugin's pyproject.toml
[project.entry-points."sunstone.plugins"]
my-plugin = "my_package:MyPlugin"
Plugin Configuration¶
Configuration is loaded with cascading precedence:
datasets.yaml→plugins.<name>section (highest priority)pyproject.toml→[tool.sunstone.plugins.<name>]section- Environment variables →
SUNSTONE_PLUGIN_<NAME>_<KEY>
Built-in URL Handlers¶
| Scheme | Handler | Extra |
|---|---|---|
| Local files | LocalFileHandler |
Built-in |
http://, https:// |
HttpURLHandler |
Built-in (with SSRF protection) |
gs:// |
GcsURLHandler |
Requires sunstone-py[gcs] |
s3://, r2:// |
S3URLHandler |
Requires sunstone-py[s3] |
Unit-Aware Arithmetic¶
sunstone-py integrates with Pint for unit-aware column arithmetic.
Unit Modes¶
Set via SUNSTONE_UNIT_MODE environment variable or programmatically:
from sunstone.units import set_unit_mode
set_unit_mode('strict') # Raise on unit mismatch
set_unit_mode('auto') # Auto-convert compatible units
set_unit_mode('relaxed') # No unit validation (default)
| Mode | Add/Sub mismatch | Mul/Div | Unknown units |
|---|---|---|---|
relaxed |
Allowed | Allowed | Allowed |
strict |
Error | Computes result unit | Error |
auto |
Auto-converts if compatible | Computes result unit | Warning |
Setting Units on Columns¶
Unit Tracking Through Operations¶
When columns with units are used in merge, join, or concat operations, sunstone validates unit compatibility and (in auto mode) applies conversions automatically.
QUDT Round-Tripping¶
Units stored as QUDT URIs in datasets.yaml are preserved through read/write cycles via the unit_source field on FieldSchema.
RDF Value Wrappers¶
Plain Python literals (str, int, float, bool, datetime, Decimal)
cover most of what shows up in Metadata.custom_properties. These
three wrappers disambiguate the cases Python's type system cannot
distinguish on its own.
IRI¶
A subclass of str flagged as an IRI reference. Stays string-comparable
and JSON-friendly, but isinstance(x, IRI) distinguishes it from a
plain string literal. Prefix resolution (e.g. sosa:NDVI → full URI)
runs at JSON-LD serialisation time against Metadata.rdf_prefixes.
LangString¶
A language-tagged literal. Serialises to JSON-LD as
{"@value": ..., "@language": ...}.
Attributes:
value(str)lang(str): BCP-47 tag, e.g."en","fr-CA".
TypedLiteral¶
A literal with an explicit XSD datatype. Use when Python-type inference
would pick the wrong xsd type. Serialises to JSON-LD as
{"@value": ..., "@type": ...}.
Attributes:
value(Any)datatype(str): e.g."xsd:double".
Constants¶
STANDARD_RDF_PREFIXES¶
Dictionary of built-in RDF prefix bindings that are always available in datasets.yaml and the generated datapackage.json without needing to be declared.
from sunstone import STANDARD_RDF_PREFIXES
print(STANDARD_RDF_PREFIXES['si'])
# 'https://sunstone.institute/rdf/vocab#'
See RDF Prefixes in datasets.yaml for the full table and usage.
License Compatibility¶
from sunstone.licenses import (
is_valid_spdx,
known_licenses,
get_properties,
check_compatibility,
get_most_restrictive_license,
derive_compatible_target,
LicenseProperties,
LicenseCompatibilityResult,
LicenseCompatibilityError,
)
Programmatic counterparts to the sunstone license CLI. The rules engine and embedded registry are documented in Concepts → License Compatibility.
is_valid_spdx(identifier)¶
True if identifier is a recognised SPDX id, an entry in the embedded registry, or any well-formed LicenseRef-* identifier per the SPDX spec. This is the validator used by lint rule R006.
Parameters:
identifier(str): License identifier to validate.
Returns: bool
known_licenses()¶
Return every license in the embedded registry, sorted by SPDX identifier.
Returns: list[LicenseProperties]
get_properties(identifier)¶
Return the LicenseProperties for identifier, or None if it is not in the embedded registry. Returns None even for valid LicenseRef-* ids that are not registered — callers must decide how to treat unknown licenses.
Parameters:
identifier(str): SPDX identifier or alias.
Returns: LicenseProperties | None
check_compatibility(source_licenses, target_license)¶
Check whether target_license is compatible with every license in source_licenses. This is the function the CLI and the writers' check_license enforcement both call.
Parameters:
source_licenses(Iterable[str]): SPDX identifiers of source licenses.target_license(str): Proposed target license identifier.
Returns: LicenseCompatibilityResult
Example:
from sunstone.licenses import check_compatibility
result = check_compatibility(
source_licenses=["CC-BY-NC-4.0", "CC-BY-4.0"],
target_license="CC-BY-4.0",
)
if not result.compatible:
for conflict in result.conflicts:
print(conflict)
if result.suggestions:
print("Try one of:", result.suggestions)
get_most_restrictive_license(licenses)¶
Return the SPDX identifier of the most restrictive license in licenses, or None if none of them are registered. Useful for suggesting a target license for a derived dataset.
Ordering, most-to-least restrictive: ShareAlike > NonCommercial > Attribution > Public Domain.
Parameters:
licenses(Iterable[str]): SPDX identifiers to compare.
Returns: str | None
derive_compatible_target(source_licenses)¶
Derive a target license that satisfies every source. Used by the writers to auto-assign an output license when none has been declared:
- A single unique source license is returned as-is (the output inherits it).
- Multiple unique source licenses produce the most restrictive registry license that satisfies every source.
- Returns
Nonewhen no compatible target exists — mutually incompatible ShareAlike families, or unknown identifiers among multiple sources that prevent verification.
Parameters:
source_licenses(Iterable[str]): SPDX identifiers of source licenses.
Returns: str | None
LicenseProperties¶
Frozen dataclass describing a license in the registry.
Attributes:
spdx(str): SPDX identifier (e.g.'CC-BY-4.0').name(str): Human-readable name.public_domain(bool): True for CC0, PDDL, and similar — compatible with everything.attribution(bool): Downstream users must credit the source.share_alike(bool): Derivatives must use the same (same-family) license.non_commercial(bool): Commercial use is forbidden downstream.family(str | None): ShareAlike family identifier — only same-family SA licenses are mutually compatible.aliases(tuple[str, ...]): Alternate identifiers (case-insensitive) that map to this license.
LicenseCompatibilityResult¶
Outcome of a check_compatibility call.
Attributes:
target(str): The proposed target license identifier.sources(list[str]): Source identifiers considered (deduplicated, preserving order).compatible(bool): True iftargetsatisfies every source's downstream requirements.conflicts(list[str]): Human-readable conflict descriptions; empty whencompatibleisTrue.suggestions(list[str]): Target licenses that would satisfy every known source (best-effort).unknown_sources(list[str]): Source identifiers not present in the registry — excluded from the rule check.unknown_target(bool): True iftargetis not in the registry; the result then carries a conflict explaining that compatibility could not be verified.
LicenseCompatibilityError¶
Subclass of SunstoneError. Raised by to_csv() / to_parquet() when check_license=True and the effective target license conflicts with a source license. See Error handling → LicenseCompatibilityError for the recovery pattern.
Environment Configuration¶
Sunstone resolves environment settings from cascading TOML files and
overlays them on os.environ. Within a single environment definition,
field-level merging follows the same precedence: project > user >
system. See the sunstone env commands
for the CLI surface.
File precedence (highest wins for active-environment selection):
SUNSTONE_DATA_ENVenv var..sunstone/data_platform.toml(project, walked up from cwd).~/.config/sunstone/data_platform.toml(user)./etc/sunstone/data_platform.toml(system).
Environment¶
Resolved environment configuration.
Attributes:
name(str): Active environment name.source(str): The file (orSUNSTONE_DATA_ENV) that selected this environment.vars(Mapping[str, str]): Flattened key/value pairs from top-level scalars and plugin-namespaced subtables. Keys are uppercased and hyphens become underscores.sections(Mapping[str, Any]): Typed models built by registeredEnvSectionProviderplugins, keyed by section name.
Methods:
activate() -> dict[str, str]: Layervarsontoos.environ. Pre-existing real env vars always win — this never overwrites them. Returns the dict of keys this call actually set.
activate_environment(*, system_config=None, user_config=None, project_config=None)¶
Convenience: resolve the active environment and call .activate().
Returns: dict[str, str] — keys actually set in os.environ
(empty when no active environment is configured or all keys were
already set).
Example:
import sunstone
applied = sunstone.activate_environment()
print(f"Applied {len(applied)} env vars from active environment")
resolve_environment(*, system_config=None, user_config=None, project_config=None)¶
Load all config files, merge environments, resolve the active name,
flatten keys, resolve op:// credential references, and build typed
section models for any registered EnvSectionProvider plugins.
Returns: Environment | None — None if no active environment is configured.
Raises: ValueError if the active environment name does not match
any defined environment, or a section model fails validation.
DataEnvironment (deprecated)¶
Deprecated alias retained for backward compatibility. Use
Environment instead.
Exceptions¶
from sunstone.exceptions import (
SunstoneError,
DatasetNotFoundError,
StrictModeError,
DatasetValidationError,
LineageError,
)
from sunstone.errors import IncompatibleAssetKindError
from sunstone.licenses import LicenseCompatibilityError
SunstoneError¶
Base exception for all sunstone-py errors.
DatasetNotFoundError¶
Raised when dataset not found in datasets.yaml.
Example:
try:
df = pd.read_csv('missing.csv', project_path=PROJECT_PATH)
except DatasetNotFoundError as e:
print(f"Dataset not registered: {e}")
StrictModeError¶
Raised when operation blocked in strict mode.
Example:
try:
df.to_csv('new.csv', slug='new', name='New', strict=True)
except StrictModeError as e:
print(f"Strict mode violation: {e}")
DatasetValidationError¶
Raised when dataset validation fails.
LineageError¶
Raised when lineage tracking encounters an error.
Type Hints¶
sunstone-py includes complete type hints for IDE support: