shillelagh.backends.apsw.dialects package

Submodules

shillelagh.backends.apsw.dialects.base module

A SQLALchemy dialect.

class shillelagh.backends.apsw.dialects.base.APSWDialect(adapters: List[str] | None = None, adapter_kwargs: Dict[str, Dict[str, Any]] | None = None, safe: bool = False, **kwargs: Any)

Bases: SQLiteDialect

A SQLAlchemy dialect for Shillelagh.

The dialect is based on the SQLiteDialect, since we’re using APSW.

colspecs: Dict[TypeEngine, TypeEngine] = {}

A dictionary of TypeEngine classes from sqlalchemy.types mapped to subclasses that are specific to the dialect class. This dictionary is class-level only and is not accessed from the dialect instance itself.

create_connect_args(url: URL) → Tuple[Tuple, Dict[str, Any]]

Build DB-API compatible connection arguments.

Given a URL object, returns a tuple consisting of a (*args, **kwargs) suitable to send directly to the dbapi’s connect function. The arguments are sent to the Dialect.connect() method which then runs the DBAPI-level connect() function.

The method typically makes use of the URL.translate_connect_args() method in order to generate a dictionary of options.

The default implementation is:

def create_connect_args(self, url):
    opts = url.translate_connect_args()
    opts.update(url.query)
    return ([], opts)

Parameters:

url – a URL object

Returns:

a tuple of (*args, **kwargs) which will be passed to the Dialect.connect() method.

See also

URL.translate_connect_args()

classmethod dbapi()

Return the DB API module.

do_ping(dbapi_connection: _ConnectionFairy) → bool

Return true if the database is online.

driver: str = 'apsw'

identifying name for the dialect’s DBAPI

get_columns(connection: _ConnectionFairy, table_name: str, schema: str | None = None, **kwargs: Any) → List[SQLAlchemyColumn]

Return information about columns in table_name.

Given a _engine.Connection, a string table_name, and an optional string schema, return column information as a list of dictionaries corresponding to the ReflectedColumn dictionary.

This is an internal dialect method. Applications should use Inspector.get_columns().

has_table(connection: _ConnectionFairy, table_name: str, schema: str | None = None, info_cache: Dict[Any, Any] | None = None, **kwargs: Any) → bool

Return true if a given table exists.

name: str = 'shillelagh'

identifying name for the dialect from a DBAPI-neutral point of view (i.e. ‘sqlite’)

supports_sane_rowcount: bool = False

Indicate whether the dialect properly implements rowcount for UPDATE and DELETE statements.

supports_statement_cache: bool = True

indicates if this dialect supports caching.

All dialects that are compatible with statement caching should set this flag to True directly on each dialect class and subclass that supports it. SQLAlchemy tests that this flag is locally present on each dialect subclass before it will use statement caching. This is to provide safety for legacy or new dialects that are not yet fully tested to be compliant with SQL statement caching.

New in version 1.4.5.

See also

engine_thirdparty_caching

class shillelagh.backends.apsw.dialects.base.SQLAlchemyColumn

Bases: TypedDict

A custom type for a SQLAlchemy column.

autoincrement: str

default: str | None

name: str

nullable: bool

primary_key: int

type: TypeEngine

shillelagh.backends.apsw.dialects.base.get_adapter_for_table_name(connection: _ConnectionFairy, table_name: str) → Adapter

Return an adapter associated with a connection.

This function instantiates the adapter responsible for a given table name, using the connection to properly pass any adapter kwargs.

shillelagh.backends.apsw.dialects.gsheets module

shillelagh.backends.apsw.dialects.safe module

A “safe” Shillelagh dialect.

When this dialect is used only adapters marked as safe and explicitly listed are loaded.

class shillelagh.backends.apsw.dialects.safe.APSWSafeDialect(adapters: List[str] | None = None, adapter_kwargs: Dict[str, Dict[str, Any]] | None = None, **kwargs: Any)

Bases: APSWDialect

A “safe” Shillelagh dialect.

This dialect can be used with the shillelagh+safe:// URI:

>>> from sqlalchemy.engine import create_engine
>>> engine = create_engine("shillelagh+safe://", adapters=["socrata"])

The dialect only loads the adapters explicitly listed (“socrata”, in the example above), and only if they’re marked as safe.

create_connect_args(url: URL) → Tuple[Tuple, Dict[str, Any]]

Build DB-API compatible connection arguments.

Given a URL object, returns a tuple consisting of a (*args, **kwargs) suitable to send directly to the dbapi’s connect function. The arguments are sent to the Dialect.connect() method which then runs the DBAPI-level connect() function.

The method typically makes use of the URL.translate_connect_args() method in order to generate a dictionary of options.

The default implementation is:

def create_connect_args(self, url):
    opts = url.translate_connect_args()
    opts.update(url.query)
    return ([], opts)

Parameters:

url – a URL object

Returns:

a tuple of (*args, **kwargs) which will be passed to the Dialect.connect() method.

See also

URL.translate_connect_args()

supports_statement_cache: bool = True

indicates if this dialect supports caching.

All dialects that are compatible with statement caching should set this flag to True directly on each dialect class and subclass that supports it. SQLAlchemy tests that this flag is locally present on each dialect subclass before it will use statement caching. This is to provide safety for legacy or new dialects that are not yet fully tested to be compliant with SQL statement caching.

New in version 1.4.5.

See also

engine_thirdparty_caching

Module contents