shillelagh.adapters package

Subpackages

Submodules

shillelagh.adapters.base module

Base class for adapters.

class shillelagh.adapters.base.Adapter(*args: Any, **kwargs: Any)[source]

Bases: object

An adapter to a table.

Adapters provide an interface to resources, so they can be queried via SQL. An adapter instance represents a virtual table, and the adapter is responsible for fetching data and metadata from the resource, and possibly insert, delete, or update rows.

In order to find an adapter responsible for a given table name, adapters need to be registered under the “shillelagh.adapter” entry point, eg:

# setup.cfg
[options.entry_points]
shillelagh.adapter =
    custom_adapter = shillelagh.adapters.api.custom:CustomAdapter

Adapters also need to implement the supports method. Given a table name, the method should return true if the table is supported by the adapter.

close() → None[source]

Close the adapter.

Adapters should use this method to perform any pending changes when the connection is closed.

delete_data(row_id: int) → None[source]

Delete a row from the table.

delete_row(row_id: int) → None[source]

Delete a row from the table.

This method is identical to delete_data, only here for symmetry.

drop_table() → None[source]

Drop a table.

get_columns() → Dict[str, Field][source]

Return the columns available in the table.

This method is called for every query, so make sure it’s cheap. For most (all?) tables this won’t change, so you can store it in an instance attribute.

get_cost(filtered_columns: List[Tuple[str, Operator]], order: List[Tuple[str, Literal[Order.ASCENDING] | Literal[Order.DESCENDING]]]) → float[source]

Estimate the query cost.

The base adapter returns a fixed cost, and custom adapter can implement their own cost estimation.

get_data(bounds: Dict[str, Filter], order: List[Tuple[str, Literal[Order.ASCENDING] | Literal[Order.DESCENDING]]], **kwargs: Any) → Iterator[Dict[str, Any]][source]

Yield rows as adapter-specific types.

This method expects rows to be in the storage format. Eg, for the CSV adapter datetime columns would be stored (and yielded) as strings. The get_rows method will use the adapter fields to convert these values into native Python types (in this case, a proper datetime.datetime).

Missing values (NULLs) may be omitted from the dictionary; they will be replaced by None by the backend.

get_metadata() → Dict[str, Any][source]

Return any extra metadata about the table.

get_rows(bounds: Dict[str, Filter], order: List[Tuple[str, Literal[Order.ASCENDING] | Literal[Order.DESCENDING]]], **kwargs: Any) → Iterator[Dict[str, Any]][source]

Yield rows as native Python types.

insert_data(row: Dict[str, Any]) → int[source]

Insert a single row with adapter-specific types.

The rows will be formatted according to the adapter fields. Eg, if an adapter represents timestamps as ISO strings, and timestamp values will be ISO strings.

insert_row(row: Dict[str, Any]) → int[source]

Insert a single row with native Python types.

The row types will be converted to the native adapter types, and passed to insert_data.

static parse_uri(uri: str) → Tuple[Any, ...][source]

Parse table name, and return arguments to instantiate adapter.

safe = False
static supports(uri: str, fast: bool = True, **kwargs: Any) → bool | None[source]

Return if a given table is supported by the adapter.

The discovery is done in 2 passes. First all adapters have their methods called with fast=True. On the first pass adapters should implement a cheap method, without any network calls.

If no adapter returns True a second pass is made with fast=False using only adapters that returned None on the first pass. In this second pass adapters can perform network requests to get more information about the URI.

The method receives the table URI, as well as the adapter connection arguments, eg:

>>> from shillelagh.backends.apsw.db import connect
>>> connection = connect(
...     ':memory:',
...     adapter_kwargs={"gsheetsapi": {"catalog":
...         {"table": "https://docs.google.com/spreadsheets/d/1/"}}},
... )

This would call all adapters in order to find which one should handle the table table. The Gsheets adapter would be called with:

>>> from shillelagh.adapters.api.gsheets.adapter import GSheetsAPI
>>> GSheetsAPI.supports("table", fast=True,  # first pass
...     catalog={"table": "https://docs.google.com/spreadsheets/d/1"})
True
supports_limit = False
supports_offset = False
supports_requested_columns = False
update_data(row_id: int, row: Dict[str, Any]) → None[source]

Update a single row with adapter-specific types.

This method by default will call a delete followed by an insert. Adapters can implement their own more efficient methods.

update_row(row_id: int, row: Dict[str, Any]) → None[source]

Update a single row with native Python types.

shillelagh.adapters.registry module

Registry for adapters.

Inspired by SQLAlchemy’s PluginLoader.

class shillelagh.adapters.registry.AdapterLoader[source]

Bases: object

Adapter registry, allowing new adapters to be registered.

add(name: str, adapter: Type[Adapter]) → None[source]

Add an adapter class directly.

clear() → None[source]

Remove all registered adapters.

load(name: str, safe: bool = False, warn: bool = False) → Type[Adapter][source]

Load a given entry point by its name.

load_all(adapters: List[str] | None = None, safe: bool = False) → Dict[str, Type[Adapter]][source]

Load all the adapters given a list of names.

If safe is True all adapters must be safe and present in the list of names. Otherwise adapters can be unsafe, and if the list is None everything is returned.

register(name: str, modulepath: str, classname: str) → None[source]

Register a new adapter.

exception shillelagh.adapters.registry.UnsafeAdaptersError[source]

Bases: InterfaceError

Raised when multiple adapters have the same name.

Module contents