"""Authoring surface for project-level datasources."""
from __future__ import annotations
import inspect
import re
from contextvars import ContextVar
from dataclasses import MISSING, Field, dataclass, field, fields
from typing import Any, ClassVar, TypeAlias, cast, get_args, get_origin
from marivo._authoring.model import AuthoringContract
from marivo.datasource.errors import (
DatasourceFieldInvalidError,
DatasourceSecretInPlaintextError,
repair,
)
from marivo.datasource.ir import AiContextIR, DatasourceIR, DatasourceSourceLocation
from marivo.datasource.typing import AiContextValue
from marivo.refs import DatasourceKind, Ref, SemanticKind, _validate_segment
from marivo.refs import ref as ref_factory
def _build_ai_context(ai_context: AiContextValue | None) -> AiContextIR:
if ai_context is None:
return AiContextIR()
if not isinstance(ai_context, AiContextValue):
raise DatasourceFieldInvalidError(
message=(
"ai_context= expects an AiContextValue from ms.ai_context(...), "
"not a raw dict. Construct it explicitly with "
"ms.ai_context(business_definition=..., guardrails=[...]). "
"summary= and other unsupported metadata keys are not accepted."
),
expected="an AiContextValue from ms.ai_context(...)",
received=type(ai_context).__name__,
location="datasource ai_context",
repair=repair(
kind="reauthor",
canonical_id="duckdb",
action="Construct ai_context with ms.ai_context(...).",
),
)
return AiContextIR(
business_definition=ai_context.business_definition,
guardrails=ai_context.guardrails,
)
SENSITIVE_FIELD_STEMS = frozenset(
{
"user",
"password",
"token",
"secret",
"secret_key",
"access_key",
"auth",
"private_key",
"passphrase",
"api_key",
}
)
JsonValue: TypeAlias = str | int | float | bool | list["JsonValue"] | dict[str, "JsonValue"] | None
_META_FIELDS = frozenset({"name", "ai_context", "extra", "fields", "env_refs"})
def _description(text: str) -> dict[str, str]:
return {"description": text}
def _is_field_required(dataclass_field: Field[object]) -> bool:
return dataclass_field.default is MISSING and dataclass_field.default_factory is MISSING
def _allows_none(annotation: object) -> bool:
"""Check whether a type annotation admits None.
Handles both resolved type objects and string annotations
(from ``from __future__ import annotations``).
"""
if annotation is None or annotation is type(None):
return True
if isinstance(annotation, str):
return "None" in annotation and "|" in annotation
origin = get_origin(annotation)
if origin is None:
return False
return type(None) in get_args(annotation)
def _is_jsonable_value(value: Any) -> bool:
if isinstance(value, str | int | float | bool) or value is None:
return True
if isinstance(value, list):
return all(_is_jsonable_value(item) for item in value)
if isinstance(value, dict):
return all(isinstance(key, str) and _is_jsonable_value(item) for key, item in value.items())
return False
def _normalize_value(value: object) -> JsonValue:
if isinstance(value, tuple):
return [_normalize_value(item) for item in value]
if isinstance(value, list):
return [_normalize_value(item) for item in value]
if isinstance(value, dict):
return {key: _normalize_value(item) for key, item in value.items()}
return cast("JsonValue", value)
def _validate_jsonable_field(name: str, key: str, value: object) -> JsonValue:
normalized = _normalize_value(value)
if not _is_jsonable_value(normalized):
raise DatasourceFieldInvalidError(
message=(
f"datasource {name!r} field {key!r} has unsupported value type "
f"{type(value).__name__}"
),
expected="a JSON-compatible datasource field value",
received=type(value).__name__,
location=f"models/datasources/ entry {name!r} field {key!r}",
repair=repair(
kind="reauthor",
canonical_id="duckdb",
action="Use a JSON-compatible datasource field value.",
),
)
return normalized
def _http_headers_env(name: str, value: object) -> dict[str, str]:
if not isinstance(value, dict) or not value:
raise DatasourceFieldInvalidError(
message=f"datasource {name!r} http_headers_env must be a non-empty mapping",
expected="HTTP header names mapped to environment variable names",
received=repr(value),
location=f"datasource {name!r} HTTP auth",
repair=repair(
kind="reauthor",
canonical_id="duckdb",
action="Map each required HTTP header to one secret environment variable.",
),
)
normalized: dict[str, str] = {}
for header_name, env_var in value.items():
if not isinstance(header_name, str) or not re.fullmatch(
r"[!#$%&'*+\-.^_`|~0-9A-Za-z]+", header_name
):
raise DatasourceFieldInvalidError(
message="DuckDB custom HTTP authentication header name is invalid",
expected="a non-empty HTTP header token",
received=repr(header_name),
location=f"datasource {name!r} HTTP auth",
repair=repair(
kind="reauthor",
canonical_id="duckdb",
action="Use valid header names such as x-secretid and x-signature.",
),
)
if not isinstance(env_var, str) or not env_var:
raise DatasourceFieldInvalidError(
message="DuckDB custom HTTP header environment reference is invalid",
expected="a non-empty environment variable name",
received=repr(env_var),
location=f"datasource {name!r} HTTP auth header {header_name!r}",
repair=repair(
kind="environment",
canonical_id="duckdb",
action="Reference a non-empty environment variable name.",
),
)
normalized[header_name] = env_var
return normalized
@dataclass
class DatasourceLoaderContext:
pending_objects: list[DatasourceIR] = field(default_factory=list)
@dataclass(frozen=True, kw_only=True)
class _SpecBase:
"""Shared behavior for concrete datasource specification classes."""
name: str = field(
metadata=_description(
"Global datasource name; letters, digits, underscores, and hyphens only."
)
)
ai_context: AiContextValue | None = field(
default=None,
metadata=_description(
"Optional AI-facing datasource context, via ms.ai_context(...). "
"Put text descriptions in ai_context.business_definition."
),
)
extra: dict[str, JsonValue] | None = field(
default=None,
metadata=_description(
"Rare JSON-safe ibis keyword arguments not modeled by the typed class."
),
)
backend_type: ClassVar[str]
fields: dict[str, JsonValue] = field(init=False)
env_refs: dict[str, str] = field(init=False)
@property
def ref(self) -> Ref[DatasourceKind]:
"""Typed reference to this datasource for semantic authoring."""
return ref_factory.datasource(self.name)
def __post_init__(self) -> None:
validate_datasource_name(self.name)
self._validate_required_string_fields()
literal_fields, env_refs = self._split_declared_fields()
object.__setattr__(self, "fields", literal_fields)
object.__setattr__(self, "env_refs", env_refs)
object.__setattr__(self, "ai_context", _build_ai_context(self.ai_context))
def contract(self) -> AuthoringContract:
"""Return the mechanical registration contract for this declaration."""
from marivo.datasource._capabilities.contracts import contract_for_spec
return contract_for_spec(self.name)
def _validate_required_string_fields(self) -> None:
for dataclass_field in fields(self):
if dataclass_field.name in _META_FIELDS:
continue
value = getattr(self, dataclass_field.name)
if value is None:
continue
if _is_field_required(dataclass_field) and isinstance(value, str) and not value:
raise DatasourceFieldInvalidError(
message=(
f"datasource {self.name!r} field {dataclass_field.name!r} "
"must be a non-empty string"
),
expected="a non-empty string",
received=repr(value),
location=f"models/datasources/ entry {self.name!r} field {dataclass_field.name!r}",
repair=repair(
kind="reauthor",
canonical_id="duckdb",
action="Provide a non-empty required datasource field.",
),
)
def _split_declared_fields(self) -> tuple[dict[str, JsonValue], dict[str, str]]:
literal_fields: dict[str, JsonValue] = {}
env_refs: dict[str, str] = {}
for dataclass_field in fields(self):
key = dataclass_field.name
if key in _META_FIELDS:
continue
value = getattr(self, key)
if value is None and _allows_none(dataclass_field.type):
continue
if key == "http_headers_env":
for header_name, env_var in _http_headers_env(self.name, value).items():
env_refs[f"http_header:{header_name}"] = env_var
continue
if key.endswith("_env"):
if value is None:
continue
stem = key[: -len("_env")]
if not isinstance(value, str) or not value:
raise DatasourceFieldInvalidError(
message=f"datasource {self.name!r} field {key!r} must be a non-empty env var name",
expected="a non-empty environment variable name",
received=repr(value),
location=f"models/datasources/ entry {self.name!r} field {key!r}",
repair=repair(
kind="reauthor",
canonical_id="duckdb",
action="Reference a non-empty environment variable name.",
),
)
env_refs[stem] = value
continue
literal_fields[key] = _validate_jsonable_field(self.name, key, value)
if self.extra:
for key, value in self.extra.items():
if key in SENSITIVE_FIELD_STEMS:
raise DatasourceSecretInPlaintextError(
message=(
f"datasource {self.name!r} field {key!r} is sensitive and must not "
"be stored as a literal"
),
expected="an environment-variable reference for a sensitive field",
received=key,
location=f"models/datasources/ entry {self.name!r} field {key!r}",
repair=repair(
kind="environment",
canonical_id="duckdb",
action="Use the matching *_env datasource field.",
snippet=f'{key}_env="<ENV_VAR>"',
),
)
literal_fields[key] = _validate_jsonable_field(self.name, key, value)
return literal_fields, env_refs
def to_ir(self, *, location: DatasourceSourceLocation) -> DatasourceIR:
return DatasourceIR(
semantic_id=self.name,
name=self.name,
backend_type=self.backend_type,
fields=dict(self.fields),
env_refs=dict(self.env_refs),
ai_context=cast("AiContextIR", self.ai_context),
python_symbol=self.name,
location=location,
)
[docs]
@dataclass(frozen=True, kw_only=True)
class DuckDBSpec(_SpecBase):
"""DuckDB datasource specification."""
backend_type: ClassVar[str] = "duckdb"
path: str = field(
default=":memory:", metadata=_description("DuckDB database path; defaults to in-memory.")
)
read_only: bool = field(
default=False, metadata=_description("Open the DuckDB database in read-only mode.")
)
http_scope: str | None = field(
default=None,
metadata=_description("Optional HTTP(S) URL prefix for scoped remote JSON auth."),
)
http_bearer_token_env: str | None = field(
default=None,
metadata=_description("Environment variable for a scoped HTTP bearer token."),
)
http_headers_env: dict[str, str] | None = field(
default=None,
metadata=_description(
"Optional custom HTTP header names mapped to secret environment variables."
),
)
def __post_init__(self) -> None:
super().__post_init__()
has_bearer = self.http_bearer_token_env is not None
has_custom_headers = self.http_headers_env is not None
if has_bearer and has_custom_headers:
raise DatasourceFieldInvalidError(
message="DuckDB HTTP auth accepts bearer token or custom headers, not both",
expected="one HTTP authentication mode",
received="bearer and custom headers",
location=f"datasource {self.name!r} HTTP auth",
repair=repair(
kind="reauthor",
canonical_id="duckdb",
action="Keep either http_bearer_token_env or http_headers_env.",
),
)
if (has_bearer or has_custom_headers) and self.http_scope is None:
raise DatasourceFieldInvalidError(
message="DuckDB HTTP auth requires an explicit URL scope",
expected="an http(s) http_scope prefix",
received="missing http_scope",
location=f"datasource {self.name!r} HTTP auth",
repair=repair(
kind="reauthor",
canonical_id="duckdb",
action="Set http_scope to the narrow API URL prefix.",
),
)
if self.http_scope is not None and not (has_bearer or has_custom_headers):
raise DatasourceFieldInvalidError(
message="DuckDB HTTP auth scope requires an authentication declaration",
expected="http_bearer_token_env or http_headers_env",
received="http_scope without authentication",
location=f"datasource {self.name!r} HTTP auth",
repair=repair(
kind="reauthor",
canonical_id="duckdb",
action="Remove http_scope or declare one environment-backed auth mode.",
),
)
if self.http_scope is not None and not re.match(r"^https?://", self.http_scope, re.I):
raise DatasourceFieldInvalidError(
message="DuckDB HTTP auth scope must be an HTTP(S) URL prefix",
expected="http_scope beginning with http:// or https://",
received=self.http_scope,
location=f"datasource {self.name!r} HTTP auth",
repair=repair(
kind="reauthor",
canonical_id="duckdb",
action="Use an explicit HTTP(S) URL prefix.",
),
)
[docs]
@dataclass(frozen=True, kw_only=True)
class SQLiteSpec(_SpecBase):
"""SQLite datasource specification."""
backend_type: ClassVar[str] = "sqlite"
path: str = field(
default=":memory:", metadata=_description("SQLite database path; defaults to in-memory.")
)
read_only: bool = field(
default=False, metadata=_description("Open the SQLite database in query-only mode.")
)
type_map: dict[str, str] | None = field(
default=None,
metadata=_description(
"Optional mapping from declared SQLite type names to Ibis type strings."
),
)
[docs]
@dataclass(frozen=True, kw_only=True)
class TrinoSpec(_SpecBase):
"""Trino datasource specification."""
backend_type: ClassVar[str] = "trino"
host: str = field(metadata=_description("Trino coordinator host."))
catalog: str = field(
metadata=_description("Trino catalog; mapped to ibis database at connect time.")
)
port: int | None = field(
default=None, metadata=_description("Trino port; ibis default is 8080.")
)
schema: str | None = field(default=None, metadata=_description("Optional default schema."))
source: str | None = field(
default=None, metadata=_description("Optional client application/source tag.")
)
timezone: str | None = field(
default=None, metadata=_description("Optional engine session timezone.")
)
http_scheme: str | None = field(default=None, metadata=_description("Set to 'https' for TLS."))
client_tags: tuple[str, ...] | None = field(
default=None, metadata=_description("Optional Trino client tags.")
)
session_properties: dict[str, JsonValue] | None = field(
default=None, metadata=_description("Optional Trino session properties.")
)
user_env: str | None = field(
default=None, metadata=_description("Environment variable for Trino user.")
)
auth_env: str | None = field(
default=None,
metadata=_description("Environment variable for Trino auth token or password."),
)
[docs]
@dataclass(frozen=True, kw_only=True)
class MySQLSpec(_SpecBase):
"""MySQL datasource specification."""
backend_type: ClassVar[str] = "mysql"
host: str = field(metadata=_description("MySQL host."))
database: str = field(metadata=_description("MySQL database name."))
port: int | None = field(
default=None, metadata=_description("MySQL port; ibis default is 3306.")
)
autocommit: bool | None = field(
default=None, metadata=_description("Optional autocommit override.")
)
user_env: str | None = field(
default=None, metadata=_description("Environment variable for MySQL user.")
)
password_env: str | None = field(
default=None, metadata=_description("Environment variable for MySQL password.")
)
[docs]
@dataclass(frozen=True, kw_only=True)
class PostgresSpec(_SpecBase):
"""Postgres datasource specification."""
backend_type: ClassVar[str] = "postgres"
host: str = field(metadata=_description("Postgres host."))
database: str = field(metadata=_description("Postgres database name."))
port: int | None = field(
default=None, metadata=_description("Postgres port; ibis default is 5432.")
)
schema: str | None = field(default=None, metadata=_description("Optional default schema."))
autocommit: bool | None = field(
default=None, metadata=_description("Optional autocommit override.")
)
user_env: str | None = field(
default=None, metadata=_description("Environment variable for Postgres user.")
)
password_env: str | None = field(
default=None, metadata=_description("Environment variable for Postgres password.")
)
[docs]
@dataclass(frozen=True, kw_only=True)
class ClickHouseSpec(_SpecBase):
"""ClickHouse datasource specification."""
backend_type: ClassVar[str] = "clickhouse"
host: str = field(metadata=_description("ClickHouse host."))
port: int | None = field(
default=None,
metadata=_description("ClickHouse port; native default is 9000, secure default is 9440."),
)
database: str | None = field(
default=None, metadata=_description("ClickHouse database; ibis default is 'default'.")
)
secure: bool | None = field(default=None, metadata=_description("Enable TLS for ClickHouse."))
settings: dict[str, JsonValue] | None = field(
default=None, metadata=_description("Optional ClickHouse settings map.")
)
user_env: str | None = field(
default=None, metadata=_description("Environment variable for ClickHouse user.")
)
password_env: str | None = field(
default=None, metadata=_description("Environment variable for ClickHouse password.")
)
DatasourceSpec: TypeAlias = (
DuckDBSpec | SQLiteSpec | TrinoSpec | MySQLSpec | PostgresSpec | ClickHouseSpec
)
def _storage_name(value: str | Ref[DatasourceKind]) -> str:
"""Return the private project storage key for a validated datasource identity."""
if type(value) is Ref and value.kind is SemanticKind.DATASOURCE:
return value.path
if type(value) is str:
validate_datasource_name(value)
return value
raise TypeError(
f"datasource storage identity must be str or Ref[datasource], got {type(value).__name__}"
)
def _require_datasource_ref(
value: object,
*,
argument: str = "datasource",
) -> Ref[DatasourceKind]:
if type(value) is Ref and value.kind is SemanticKind.DATASOURCE:
return value
raise TypeError(
f"{argument} must be Ref[datasource] from a datasource spec's .ref or "
"ref_factory.datasource('warehouse'). Do not pass a bare string."
)
_DATASOURCE_CTX: ContextVar[DatasourceLoaderContext | None] = ContextVar(
"_DATASOURCE_CTX",
default=None,
)
_AUTHORING_FILE = __file__
def _caller_location() -> DatasourceSourceLocation:
try:
frame = inspect.currentframe()
if frame is not None:
caller_frame = frame.f_back
while caller_frame is not None:
if (
caller_frame.f_code.co_filename != _AUTHORING_FILE
and caller_frame.f_globals.get("__name__") != "marivo.telemetry"
):
return DatasourceSourceLocation(
file=caller_frame.f_code.co_filename,
line=caller_frame.f_lineno,
)
caller_frame = caller_frame.f_back
except AttributeError:
pass
return DatasourceSourceLocation(file="<unknown>", line=0)
def _current_ctx() -> DatasourceLoaderContext | None:
return _DATASOURCE_CTX.get()
def _require_ctx() -> DatasourceLoaderContext:
ctx = _DATASOURCE_CTX.get()
if ctx is None:
raise DatasourceFieldInvalidError(
message="md.datasource can only be called while loading models/datasources/ files",
expected="a models/datasources loader context",
received="outside loader",
location="md.datasource",
repair=repair(
kind="reauthor",
canonical_id="load",
action="Declare datasources from models/datasources/ files.",
),
)
return ctx
def validate_datasource_name(name: Any) -> None:
if not isinstance(name, str) or not name:
raise DatasourceFieldInvalidError(
message="datasource name must be a non-empty string",
expected="a non-empty datasource storage name",
received=repr(name),
location="datasource name",
repair=repair(
kind="reauthor",
canonical_id="duckdb",
action="Provide a non-empty datasource name.",
),
)
try:
_validate_segment(name, role="datasource name")
except ValueError:
suggested = re.sub(r"[^a-z0-9_]+", "_", name.lower()).strip("_")
if not suggested:
suggested = "datasource"
if not suggested[0].isalpha():
suggested = f"ds_{suggested}"
raise DatasourceFieldInvalidError(
message=f"datasource {name!r} is not a valid datasource name",
expected="[a-z][a-z0-9_]*",
received=name,
location="datasource name",
repair=repair(
kind="reauthor",
canonical_id="duckdb",
action=(
f"Rename the datasource to {suggested!r}. Its identity becomes "
f"datasource:{suggested}. Existing ~/.marivo/secrets.toml entries are "
"not renamed; cached credentials are reused only when the conventional "
"environment-variable name remains unchanged."
),
),
) from None
def _ir_from_spec(spec: DatasourceSpec, *, location: DatasourceSourceLocation) -> DatasourceIR:
return spec.to_ir(location=location)
def _declare(spec: DatasourceSpec) -> None:
"""Internal: append a spec's IR to the current loader context."""
ctx = _require_ctx()
ctx.pending_objects.append(_ir_from_spec(spec, location=_caller_location()))
def _declare_if_loading(spec: DatasourceSpec) -> None:
ctx = _current_ctx()
if ctx is not None:
ctx.pending_objects.append(_ir_from_spec(spec, location=_caller_location()))
[docs]
def duckdb(
name: str,
*,
path: str = ":memory:",
read_only: bool = False,
http_scope: str | None = None,
http_bearer_token_env: str | None = None,
http_headers_env: dict[str, str] | None = None,
ai_context: AiContextValue | None = None,
extra: dict[str, JsonValue] | None = None,
) -> DuckDBSpec:
"""Declare a DuckDB datasource.
Args:
name: Global datasource name; letters, digits, underscores, and hyphens only.
path: DuckDB database path; defaults to in-memory.
read_only: Open the DuckDB database in read-only mode.
http_scope: Optional HTTP(S) URL prefix limiting remote JSON authentication.
http_bearer_token_env: Environment variable containing a bearer token.
http_headers_env: Optional custom HTTP header names mapped to environment
variables containing their secret values.
ai_context: Optional AI-facing context, via ``ms.ai_context(...)``.
Put text descriptions in ``business_definition``.
extra: Rare JSON-safe ibis keyword arguments not modeled by the typed class.
Returns:
``DuckDBSpec`` usable with ``md.register(...)`` or ``.ref``.
Example:
>>> import marivo.datasource as md
>>> spec = md.duckdb(name="warehouse", path=":memory:")
>>> md.register(spec)
>>> md.test(spec.ref).show()
>>> md.inspect(spec.ref, md.table("orders")).show()
>>> md.inspect(spec.ref, md.parquet("data/orders/*.parquet")).show()
``md.table(...)`` selects an internal table or view in the DuckDB
datasource. ``md.parquet(...)``, ``md.csv(...)``, and ``md.json(...)``
build DuckDB file source descriptors; they are not datasource
declarations.
Constraints:
When called while loading a datasource file, the spec is automatically
declared for that project.
"""
spec = DuckDBSpec(
name=name,
path=path,
read_only=read_only,
http_scope=http_scope,
http_bearer_token_env=http_bearer_token_env,
http_headers_env=http_headers_env,
ai_context=ai_context,
extra=extra,
)
_declare_if_loading(spec)
return spec
[docs]
def sqlite(
name: str,
*,
path: str = ":memory:",
read_only: bool = False,
type_map: dict[str, str] | None = None,
ai_context: AiContextValue | None = None,
extra: dict[str, JsonValue] | None = None,
) -> SQLiteSpec:
"""Declare a SQLite datasource.
Args:
name: Global datasource name; lowercase snake_case only.
path: SQLite database path; defaults to in-memory.
read_only: Open the SQLite database in query-only mode.
type_map: Optional mapping from declared SQLite type names to Ibis type strings.
ai_context: Optional AI-facing context, via ``ms.ai_context(...)``.
Put text descriptions in ``business_definition``.
extra: Rare JSON-safe ibis keyword arguments not modeled by the typed class.
Returns:
``SQLiteSpec`` usable with ``md.register(...)`` or ``.ref``.
Example:
>>> import marivo.datasource as md
>>> spec = md.sqlite(name="app", path="data/app.sqlite", read_only=True)
>>> md.register(spec)
>>> md.test(spec.ref).show()
>>> md.inspect(spec.ref, md.table("orders")).show()
Constraints:
When called while loading a datasource file, the spec is automatically
declared for that project. SQLite datasources support tables and views;
Parquet, CSV, and JSON file descriptors require DuckDB. SQLite does not
support median or percentile aggregation, or string strptime expressions;
use a supported aggregation and a native temporal column instead.
"""
spec = SQLiteSpec(
name=name,
path=path,
read_only=read_only,
type_map=type_map,
ai_context=ai_context,
extra=extra,
)
_declare_if_loading(spec)
return spec
[docs]
def trino(
name: str,
*,
host: str,
catalog: str,
port: int | None = None,
schema: str | None = None,
source: str | None = None,
timezone: str | None = None,
http_scheme: str | None = None,
client_tags: tuple[str, ...] | None = None,
session_properties: dict[str, JsonValue] | None = None,
user_env: str | None = None,
auth_env: str | None = None,
ai_context: AiContextValue | None = None,
extra: dict[str, JsonValue] | None = None,
) -> TrinoSpec:
"""Declare a Trino datasource.
Args:
name: Global datasource name; letters, digits, underscores, and hyphens only.
host: Trino coordinator host.
catalog: Trino catalog; mapped to ibis database at connect time.
port: Trino port; ibis default is 8080.
schema: Optional default schema.
source: Optional client application/source tag.
timezone: Optional engine session timezone.
http_scheme: Set to 'https' for TLS.
client_tags: Optional Trino client tags.
session_properties: Optional Trino session properties.
user_env: Environment variable for Trino user.
auth_env: Environment variable for Trino auth token or password.
ai_context: Optional AI-facing context, via ``ms.ai_context(...)``.
Put text descriptions in ``business_definition``.
extra: Rare JSON-safe ibis keyword arguments not modeled by the typed class.
Returns:
``TrinoSpec`` usable with ``md.register(...)`` or ``.ref``.
Example:
>>> import marivo.datasource as md
>>> spec = md.trino(
... name="warehouse",
... host="trino.example",
... catalog="hive",
... schema="analytics",
... user_env="WAREHOUSE_USER",
... auth_env="WAREHOUSE_AUTH",
... )
>>> md.register(spec)
>>> md.test(spec.ref).show()
>>> md.inspect(spec.ref, md.table("orders")).show()
Constraints:
When called while loading a datasource file, the spec is automatically
declared for that project.
Sensitive fields must use ``*_env`` references, not plaintext literals.
"""
spec = TrinoSpec(
name=name,
host=host,
catalog=catalog,
port=port,
schema=schema,
source=source,
timezone=timezone,
http_scheme=http_scheme,
client_tags=client_tags,
session_properties=session_properties,
user_env=user_env,
auth_env=auth_env,
ai_context=ai_context,
extra=extra,
)
_declare_if_loading(spec)
return spec
[docs]
def mysql(
name: str,
*,
host: str,
database: str,
port: int | None = None,
autocommit: bool | None = None,
user_env: str | None = None,
password_env: str | None = None,
ai_context: AiContextValue | None = None,
extra: dict[str, JsonValue] | None = None,
) -> MySQLSpec:
"""Declare a MySQL datasource.
Args:
name: Global datasource name; letters, digits, underscores, and hyphens only.
host: MySQL host.
database: MySQL database name.
port: MySQL port; ibis default is 3306.
autocommit: Optional autocommit override.
user_env: Environment variable for MySQL user.
password_env: Environment variable for MySQL password.
ai_context: Optional AI-facing context, via ``ms.ai_context(...)``.
Put text descriptions in ``business_definition``.
extra: Rare JSON-safe ibis keyword arguments not modeled by the typed class.
Returns:
``MySQLSpec`` usable with ``md.register(...)`` or ``.ref``.
Example:
>>> import marivo.datasource as md
>>> spec = md.mysql(
... name="oltp",
... host="mysql.example",
... database="app",
... user_env="WAREHOUSE_USER",
... password_env="WAREHOUSE_PASSWORD",
... )
>>> md.register(spec)
>>> md.test(spec.ref).show()
>>> md.inspect(spec.ref, md.table("orders", database="app")).show()
Constraints:
When called while loading a datasource file, the spec is automatically
declared for that project.
Sensitive fields must use ``*_env`` references, not plaintext literals.
"""
spec = MySQLSpec(
name=name,
host=host,
database=database,
port=port,
autocommit=autocommit,
user_env=user_env,
password_env=password_env,
ai_context=ai_context,
extra=extra,
)
_declare_if_loading(spec)
return spec
[docs]
def postgres(
name: str,
*,
host: str,
database: str,
port: int | None = None,
schema: str | None = None,
autocommit: bool | None = None,
user_env: str | None = None,
password_env: str | None = None,
ai_context: AiContextValue | None = None,
extra: dict[str, JsonValue] | None = None,
) -> PostgresSpec:
"""Declare a Postgres datasource.
Args:
name: Global datasource name; letters, digits, underscores, and hyphens only.
host: Postgres host.
database: Postgres database name.
port: Postgres port; ibis default is 5432.
schema: Optional default schema.
autocommit: Optional autocommit override.
user_env: Environment variable for Postgres user.
password_env: Environment variable for Postgres password.
ai_context: Optional AI-facing context, via ``ms.ai_context(...)``.
Put text descriptions in ``business_definition``.
extra: Rare JSON-safe ibis keyword arguments not modeled by the typed class.
Returns:
``PostgresSpec`` usable with ``md.register(...)`` or ``.ref``.
Example:
>>> import marivo.datasource as md
>>> spec = md.postgres(
... name="oltp",
... host="pg.example",
... database="app",
... user_env="WAREHOUSE_USER",
... password_env="WAREHOUSE_PASSWORD",
... )
>>> md.register(spec)
>>> md.test(spec.ref).show()
>>> md.inspect(spec.ref, md.table("orders", database="app")).show()
Constraints:
When called while loading a datasource file, the spec is automatically
declared for that project.
Sensitive fields must use ``*_env`` references, not plaintext literals.
"""
spec = PostgresSpec(
name=name,
host=host,
database=database,
port=port,
schema=schema,
autocommit=autocommit,
user_env=user_env,
password_env=password_env,
ai_context=ai_context,
extra=extra,
)
_declare_if_loading(spec)
return spec
[docs]
def clickhouse(
name: str,
*,
host: str,
port: int | None = None,
database: str | None = None,
secure: bool | None = None,
settings: dict[str, JsonValue] | None = None,
user_env: str | None = None,
password_env: str | None = None,
ai_context: AiContextValue | None = None,
extra: dict[str, JsonValue] | None = None,
) -> ClickHouseSpec:
"""Declare a ClickHouse datasource.
Args:
name: Global datasource name; letters, digits, underscores, and hyphens only.
host: ClickHouse host.
port: ClickHouse port; native default is 9000, secure default is 9440.
database: ClickHouse database; ibis default is 'default'.
secure: Enable TLS for ClickHouse.
settings: Optional ClickHouse settings map.
user_env: Environment variable for ClickHouse user.
password_env: Environment variable for ClickHouse password.
ai_context: Optional AI-facing context, via ``ms.ai_context(...)``.
Put text descriptions in ``business_definition``.
extra: Rare JSON-safe ibis keyword arguments not modeled by the typed class.
ClickHouse connections default ``autogenerate_session_id`` to ``False``
for analysis stability; pass ``{"autogenerate_session_id": True}``
only when the datasource needs ClickHouse session state such as
temporary tables.
Returns:
``ClickHouseSpec`` usable with ``md.register(...)`` or ``.ref``.
Example:
>>> import marivo.datasource as md
>>> spec = md.clickhouse(
... name="warehouse",
... host="clickhouse.example",
... database="analytics",
... user_env="WAREHOUSE_USER",
... password_env="WAREHOUSE_PASSWORD",
... )
>>> md.register(spec)
>>> md.test(spec.ref).show()
>>> md.inspect(spec.ref, md.table("orders", database="analytics")).show()
Constraints:
When called while loading a datasource file, the spec is automatically
declared for that project.
Sensitive fields must use ``*_env`` references, not plaintext literals.
"""
spec = ClickHouseSpec(
name=name,
host=host,
port=port,
database=database,
secure=secure,
settings=settings,
user_env=user_env,
password_env=password_env,
ai_context=ai_context,
extra=extra,
)
_declare_if_loading(spec)
return spec