API Reference

Every member of the Drasi class, grouped by area.

4 minute read

This is the surface of drasi.Drasi. Each member has a blocking counterpart of the same name on drasi.sync.Drasi, where the Awaitable[...] return becomes the value itself and a Stream becomes an ordinary iterator — see the blocking API.

The shapes referenced here (SourceChange, QueryResultEvent, StateStore and the rest) are declared in drasi.types, described in the typing guide.

Generated from the shipped stubs

The tables below are generated from python/drasi/_drasi.pyi, the same file your type checker and editor read, and a test fails if they fall out of step. They cannot drift from what the package exposes.

Creating an engine

Member	Returns
`id` (property)	`str`
`from_config(config: Mapping[str, Any])` (static)	`Awaitable[Drasi]`
`create(id: str, , secrets: Mapping[str, str] \| None = None, state_store: StateStore \| None = None, index_store: IndexStore \| None = None, identity: Identity \| None = None, plugins_dir: StrPath \| None = None)` (static)*	`Awaitable[Drasi]`

Lifecycle

Member	Returns
`start()`	`Awaitable[None]`
`stop()`	`Awaitable[None]`
`close()`	`Awaitable[None]`
`is_running()`	`Awaitable[bool]`

Queries

Member	Returns
add_query(self, id: str, query: str, sources: Sequence[str \| SourceSubscription], *, language: QueryLanguage = "cypher", joins: Sequence[Join] \| None = None, middleware: Sequence[Middleware] \| None = None, auto_start: bool \| None = None, enable_bootstrap: bool \| None = None, bootstrap_timeout_seconds: int \| None = None, priority_queue_capacity: int \| None = None, dispatch_buffer_capacity: int \| None = None, outbox_capacity: int \| None = None, dispatch_mode: Literal["channel", "broadcast"] \| None = None)	`Awaitable[None]`
update_query(self, id: str, query: str, sources: Sequence[str \| SourceSubscription], *, language: QueryLanguage = "cypher", joins: Sequence[Join] \| None = None, middleware: Sequence[Middleware] \| None = None, auto_start: bool \| None = None, enable_bootstrap: bool \| None = None, bootstrap_timeout_seconds: int \| None = None, priority_queue_capacity: int \| None = None, dispatch_buffer_capacity: int \| None = None, outbox_capacity: int \| None = None, dispatch_mode: Literal["channel", "broadcast"] \| None = None)	`Awaitable[None]`
`remove_query(id: str)`	`Awaitable[None]`
`start_query(id: str)`	`Awaitable[None]`
`stop_query(id: str)`	`Awaitable[None]`
`get_query_results(id: str)`	`Awaitable[list[dict[str, Any]]]`
`list_queries()`	`Awaitable[list[tuple[str, str]]]`
`get_query_status(id: str)`	`Awaitable[str]`
`wait_for_query(id: str, *, timeout: float = 30.0)`	`Awaitable[None]`

Components defined in Python

Member	Returns
`add_python_source(id: str, *, auto_start: bool = True)`	`Awaitable[None]`
`push_change(source_id: str, change: SourceChange)`	`Awaitable[None]`
`add_python_reaction(self, id: str, query_ids: Sequence[str], callback: Callable[[QueryResultEvent], object])`	`Awaitable[None]`
`add_durable_python_reaction(self, id: str, query_ids: Sequence[str], callback: Callable[[QueryResultEvent], Awaitable[object]], *, recovery_policy: Literal["strict", "auto_reset", "skip_gap"] = "strict")`	`Awaitable[None]`

Components provided by plugins

Member	Returns
`add_source(self, kind: str, id: str, config: Mapping[str, Any] \| None = None, *, auto_start: bool = True, bootstrap: Mapping[str, Any] \| None = None)`	`Awaitable[None]`
`remove_source(id: str, *, cleanup: bool = False)`	`Awaitable[None]`
`start_source(id: str)`	`Awaitable[None]`
`stop_source(id: str)`	`Awaitable[None]`
`get_source_status(id: str)`	`Awaitable[str]`
`list_sources()`	`Awaitable[list[tuple[str, str]]]`
`update_source(self, kind: str, id: str, config: Mapping[str, Any] \| None = None, *, auto_start: bool = True)`	`Awaitable[None]`
`add_reaction(self, kind: str, id: str, query_ids: Sequence[str], config: Mapping[str, Any] \| None = None, *, auto_start: bool = True)`	`Awaitable[None]`
`remove_reaction(id: str, *, cleanup: bool = False)`	`Awaitable[None]`
`start_reaction(id: str)`	`Awaitable[None]`
`stop_reaction(id: str)`	`Awaitable[None]`
`get_reaction_status(id: str)`	`Awaitable[str]`
`list_reactions()`	`Awaitable[list[tuple[str, str]]]`
`update_reaction(self, kind: str, id: str, query_ids: Sequence[str], config: Mapping[str, Any] \| None = None, *, auto_start: bool = True)`	`Awaitable[None]`

Metrics and schema

Member	Returns
`get_query_metrics(id: str)`	`Awaitable[QueryMetrics]`
`get_reaction_metrics(id: str)`	`Awaitable[dict[str, ReactionQueryMetrics]]`
`get_lifecycle_metrics()`	`Awaitable[LifecycleMetrics]`
`get_source_schema(id: str)`	`Awaitable[SourceSchema \| None]`
`get_graph_schema()`	`Awaitable[GraphSchema]`

Streaming

Member	Returns
`query_results(self, query_id: str, *, reaction_id: str \| None = None )`	`Awaitable[Stream]`
`query_events(id: str)`	`Awaitable[Stream]`
`source_events(id: str)`	`Awaitable[Stream]`
`reaction_events(id: str)`	`Awaitable[Stream]`
`all_events()`	`Awaitable[Stream]`
`query_logs(id: str)`	`Awaitable[Stream]`
`source_logs(id: str)`	`Awaitable[Stream]`
`reaction_logs(id: str)`	`Awaitable[Stream]`
`on_query_results(self, query_id: str, callback: Callable[[QueryResultEvent], object] )`	`Awaitable[None]`
`on_query_events(self, id: str, callback: Callable[[ComponentEvent], object] )`	`Awaitable[None]`
`on_source_events(self, id: str, callback: Callable[[ComponentEvent], object] )`	`Awaitable[None]`
`on_reaction_events(self, id: str, callback: Callable[[ComponentEvent], object] )`	`Awaitable[None]`
`on_all_events(callback: Callable[[ComponentEvent], object])`	`Awaitable[None]`
`on_query_logs(self, id: str, callback: Callable[[LogMessage], object] )`	`Awaitable[None]`
`on_source_logs(self, id: str, callback: Callable[[LogMessage], object] )`	`Awaitable[None]`
`on_reaction_logs(self, id: str, callback: Callable[[LogMessage], object] )`	`Awaitable[None]`

Plugins

Member	Returns
`load_plugins(self, directory: StrPath, verify: Mapping[str, str] \| None = None )`	`Awaitable[LoadSummary]`
`plugin_kinds()`	`Awaitable[PluginKinds]`
`host_info()`	`HostInfo`
`search_plugins(query: str \| None = None)`	`Awaitable[list[PluginSearchResult]]`
`list_plugin_tags(repository: str)`	`Awaitable[list[str]]`
`resolve_plugin(reference: str)`	`Awaitable[ResolvedPlugin]`
`install_plugin(self, reference: str, *, directory: StrPath \| None = None, verify: bool = False, require_signed: bool = False, trusted_identities: Sequence[tuple[str, str]] \| None = None, load: bool = True)`	`Awaitable[InstalledPlugin]`
`watch_plugins(self, directory: StrPath, *, debounce_seconds: float = 1.0 )`	`Awaitable[None]`
`pull_plugin(self, reference: str, directory: StrPath, filename: str, *, verify: bool = False, require_signed: bool = False, trusted_identities: Sequence[tuple[str, str]] \| None = None)`	`Awaitable[PulledPlugin]`
`write_lockfile(directory: StrPath)`	`Awaitable[int]`
`read_lockfile(directory: StrPath)` (static)	`list[LockedPlugin]`
`install_from_lockfile(self, directory: str, *, load: bool = True )`	`Awaitable[list[str]]`
`source_config_schema(kind: str)`	`Awaitable[ConfigSchema]`
`reaction_config_schema(kind: str)`	`Awaitable[ConfigSchema]`
`bootstrap_config_schema(kind: str)`	`Awaitable[ConfigSchema]`
`secret_store_config_schema(kind: str)`	`Awaitable[ConfigSchema]`
`use_secret_store(self, kind: str, config: Mapping[str, Any] \| None = None )`	`Awaitable[None]`

Errors

Everything raised is a DrasiError or a subclass, each carrying a stable code. See error handling for the hierarchy and the full list of codes.

Module-level members

Member	Description
`drasi.__version__`	The installed package version
`drasi.host_info()`	Versions, target triple and available index backends
`drasi.ERROR_CODES`	Every error code this build can raise
`drasi.DRASI_CORE_VERSION`	The `drasi-core` version this wheel was built against
`drasi.DRASI_LIB_VERSION`	The `drasi-lib` version this wheel was built against
`drasi.DRASI_SDK_VERSION`	The plugin SDK version this wheel was built against

Feedback

Was this page helpful?

Glad to hear it! Please tell us how we can improve.

Sorry to hear that. Please tell us how we can improve.