Install / Get started

Sqeleton can be installed using pip:

pip install sqeleton

Database drivers

To ensure that the database drivers are compatible with sqeleton, we recommend installing them along with sqeleton, using pip’s [] syntax:

pip install 'sqeleton[mysql]'
pip install 'sqeleton[postgresql]'
pip install 'sqeleton[snowflake]'
pip install 'sqeleton[presto]'
pip install 'sqeleton[oracle]'
pip install 'sqeleton[trino]'
pip install 'sqeleton[clickhouse]'
pip install 'sqeleton[vertica]'
For BigQuery, see: https://pypi.org/project/google-cloud-bigquery/

Some drivers have dependencies that cannot be installed using pip and still need to be installed manually.

It is also possible to install several databases at once. For example:

pip install 'sqeleton[mysql, postgresql]'

Note: Some shells use " for escaping instead, like:

pip install "sqeleton[mysql, postgresql]"

Connection editor

Sqeleton provides a TUI connection editor, that can be installed using:

pip install 'sqeleton[tui]'

What’s next?

Read the introduction and start coding!

Introduction to Sqeleton

Sqeleton is a Python library for querying SQL databases.

It has two complementary APIs:

Database interface, for connecting and querying.
Query builder, for constructing query expressions (ASTs).

The following sections delve into each one of them.

Database Interface

Sqeleton’s database interface enables provides a unified interface for accessing a wide array of databases.

Each database has its own dialect, which tells the compiler which SQL to produce for each expression. Extra features can be added to the dialect using mixins.

connect()

Database instances are created using the connect() function, when given a URI or equivalent dict:

>>> from sqeleton import connect
>>> connect("mysql://localhost/db")
<sqeleton.databases.mysql.MySQL object at ...>
>>> connect({"driver": "mysql", "host": "localhost", "database": "db"})
<sqeleton.databases.mysql.MySQL object at ...>

For non-cloud DBs, you can also provide the size of the local thread pool, i.e. how many worker-threads it should spawn.

Database instances have a few useful methods, like list_tables(), query_table_schema():

>>> db = connect("postgresql:///")
>>> db.name
'PostgreSQL'
>>> db.list_tables('R%') 
[('Rating',), ('Rating_del1',), ('Rating_del1p',), ('Rating_update001p',), ('Rating_update1',), ('Rating_update1p',), ('Rating_update50p',)]
>>> db.close()  
>>> db.is_closed
True

Database.query()

The query function accepts either raw SQL, or a query expression, created by the query-builder:

>>> db = connect("duckdb://:memory:")

# Query raw SQL
>>> db.query('select 1 + 1') 
[(2,)]

# Query SQL expression (more on that later)
>>> from sqeleton.queries import current_timestamp
>>> db.query(current_timestamp())
datetime.datetime(2022, 12, 28, 17, 1, 11)

It also accepts a list of statements:

from sqeleton.queries import table
>>> tmp = table('tmp', schema={'i': int})
# Query list of statements, returns the last result
>>> db.query([
...     tmp.create(),
...     tmp.insert_rows([x] for x in range(100)),
...     'select sum(i) from tmp',
... ])
[(4950,)]

It’s possible to tell the query() method the type you expect as a result, and it will validate and attempt to cast it:

# Same as default. Return a list of tuples.
>>> db.query('select 1 + 1', list)   
[(2,)]
# Return a tuple. Expects only 1 row returned.
>>> db.query('select 1 + 1', tuple)  
(2,)
# Return an int. Expects only 1 row and column returned.
>>> db.query('select 1 + 1', int) 
2

Query Builder

Sqeleton’s query-builder takes a lot of ideas from both SQLAlchemy and PyPika, however there are several notable differences.

table()

To query tables, users create a table instance, and use chain methods to build the query expression:

>>> from sqeleton import table

# SELECT * FROM tmp -- Query everything in table 'tmp'
>>> expr = table('tmp').select() 
>>> db.query(expr)
[(0,), (1,), (2,), (3,), ...]

To specify a dotted table path (for schemas, etc.), use either form:

table('foo', 'bar') compiles to foo.bar.
table(('foo', 'bar')) also compiles to foo.bar.

table attributes and `this`

In sqeleton, table attributes are accessed either through the table’s [] operator, or using the this object. When compiled, this.foo evaluates to the column of the attached table:

from sqeleton import table, this

tbl = table('tmp')

# All exprs are equivalent to: SELECT foo, bar FROM tmp
expr1 = tbl.select(tbl['foo'], tbl['bar'])
expr1 = tbl.select(this['foo'], this['bar'])
expr2 = tbl.select(this.foo, this.bar) 

It’s also possible to give this a list of attributes:

attrs = ['foo', 'bar']

# Equivalent to: SELECT foo, bar FROM tbl
expr = tbl.select(*this[attrs])

It’s recommended to prefer the this.attr syntax whenever possible, because it’s shorter, more readable, and more amenable to refactoring.

However, the [] syntax is very useful for variable attribute names.

.select(), .where(), .order_by(), .limit()

These are fairly straightforward, and similar to other libraries:

# SELECT a FROM tmp WHERE b > 10 ORDER BY c LIMIT 20
table('tmp').select(this.a).where(this.b > 10).order_by(this.c).limit(20)

There are a few things worth mentioning:

SKIP - You can provide the sqeleton.SKIP keyword to these functions, which does nothing. That can be useful for chaining conditionals. For example:

from sqeleton import SKIP

# SELECT name[, age] FROM tmp [WHERE age > 18] [LIMIT limit]
rows = (table('tmp')
        .select(this.name, (this.age if include_age else SKIP))
        .where(this.age > 18 if only_adults else SKIP)
        .limit(limit if limit is not None else SKIP)
       )

To alias columns in .select(), use keyword arguments:

# SELECT id, (first || ' ' || last) AS full_name, (age >= 18) AS is_adult FROM person 
table('person').select(
        this.id,
        full_name = this.first + " " + this.last,
        is_adult = self.age >= 18
    )

Generators

It’s possible to provide generators to select(), where(), and order_by(), enabling syntax like this:

fields = 'foo', 'bar', '_xyz'

# SELECT foo, bar FROM tmp WHERE foo > 0 AND bar > 0 AND _xyz > 0
all_above_0 = (table('tmp')
        .select(this[f] for f in fields if not f.startswith('_'))
        .where(this[f] > 0 for f in fields)
       )

.group_by(), .agg(), and .having()

Group-by in sqeleton behaves like in SQL, except for a small change in syntax:

# SELECT a, sum(b) FROM tmp GROUP BY 1
table('tmp').group_by(this.a).agg(this.b.sum())

# SELECT a, sum(b) FROM a GROUP BY 1 HAVING (b > 10)
(table('tmp')
    .group_by(this.a)
    .agg(this.b.sum())
    .having(this.b > 10)
)

These functions also accept generators and SKIP.

A call to .agg() must follow every call to group_by().

Any use of .select() will be considered a separate sub-query:

# SELECT (c + 1) FROM (SELECT b, c FROM (SELECT a FROM tmp) GROUP BY 1)
rows = (table('tmp')
        .select(this.a)
        .group_by(this.b)
        .agg(this.c)
        .select(this.c + 1)
    )

More table operations

Tables and queries also support the following methods:

# SELECT count() FROM a`
a.count()

# SELECT * FROM a UNION b
a.union(b)

# SELECT * FROM a UNION ALL b  -- concat tables
a.union_all(b)

# SELECT * FROM a EXCEPT b      -- or MINUS
a.minus(b)

# select * from a INTERSECT b
a.intersect(b)

.join(), .on()

When joining, it’s recommended to use explicit tables names, instead of this, in order to avoid potential name collisions.

person = table('person')
city = table('city')

name_and_city = (
    person
    .join(city)
    .on(person['city_id'] == city['id'])
    .select(person['id'], city['name'])
)

.on() also supports generators and SKIP.

`when()` and `.then()`

Sqeleton provides a way to construct case-when-then expressions:

from sqeleton.queries import when

# SELECT CASE
#   WHEN (type = 'text') THEN text
#   WHEN (type = 'number') THEN number
#   ELSE 'unknown type' END
# FROM foo
rows = table('foo').select(
        when(this.type == 'text').then(this.text)
        .when(this.type == 'number').then(this.number)
        .else_('unknown type')
    )

Each .when() must be followed by a .then().

Sqeleton also provides a convenience if_() function for simple conditionals:

from sqeleton import if_

# SELECT CASE WHEN b THEN c ELSE d END FROM foo
table('foo').select(if_(b, c, d))

DDL - .create(), .drop(), .truncate()

These methods create DDL expressions (or “data-definition language”).

To execute these expressions on a database, you have to call Database.query().

.create()

It’s possible to create empty new tables using a schema, or new tables populated using a query.

Both are done by using the .create() method:

source_table = table('source')

db.query([

    # CREATE TABLE new AS SELECT * FROM source
    table('new').create(source_table),

    # CREATE TABLE new AS SELECT * FROM source WHERE x > 0
    table('new_nonzero').create(source_table.where(this.x > 0)),

    # CREATE TABLE foo (id INT, name VARCHAR)
    table('foo', schema={
        id: int,
        name: str
    }).create()

])

The .create() method also accepts the following keyword parameters:

if_not_exists - Adds IF NOT EXISTS to the create statement
primary_keys - Specify primary keys when creating the table
.drop(), .truncate()

These are the simple parallels of DROP TABLE and TRUNCATE TABLE.

Add data - .insert_row(), .insert_rows(), .insert_expr()

These methods insert rows of constant values (from Python), or from a query expression.

# INSERT INTO atoms VALUES ('H', 1)
table('atoms').insert_row("H", 1)

# INSERT INTO atoms VALUES ('H', 1), ('He', 2)
rows = [
    ("H", 1),
    ("He", 2)
] 
table('atoms').insert_rows(rows)

# INSERT INTO foo SELECT * FROM bar     -- Concat 'bar' to 'foo'
table('foo').insert_expr(table('bar'))

A common pattern is to call .insert_rows() with a generator:

rows = {
    1: "H",
    2: "He",
}
table('atoms').insert_rows((sym, num) for num, sym in rows.items())

Raw SQL using `code()`

It’s possible to combine Sqeleton’s query expressions with raw SQL code.

It allows users to use features and syntax that Sqeleton doesn’t yet support.

Keep in mind that the code is very unlikely to port to other databases, so if you need to support more than one database, keep your use of code() to a minimum, and use it behind abstracting functions.

from sqeleton import code

# SELECT b, <x> FROM tmp WHERE <y>
table('tmp').select(this.b, code("<x>")).where(code("<y>"))

It’s the user’s responsibility to make sure the contents of the string given to code() are correct and safe for execution.

Strings given to code() are actually templates, and can embed query expressions given as arguments:

def tablesample(tbl, size):
    return code("{tbl} TABLESAMPLE BERNOULLI ({size})", tbl=tbl, size=size)

nonzero = table('points').where(this.x > 0, this.y > 0)

# SELECT * FROM points WHERE (x > 0) AND (y > 0) TABLESAMPLE BERNOULLI (10)
sample_expr = tablesample(nonzero)

Bound Tables & Expressions - Database.table()

Sqeleton’s query expressions are database-agnostic, which makes it easy to run the exact same queries on different databases.

While this functional style is sufficient for most purposes, it’s sometimes convenient to have a more object-oriented approach, and pass around query expressions bound to a specific database. That can be especially useful when running all the queries on the same database, or when different databases need drastically different queries.

Bound exprs support .query(), which will execute them on the bound database:

# SELECT foo FROM tmp
db.table('tmp').select(this.foo).query()

# SELECT foo FROM tmp  -- expects one row and one column
db.table('tmp').select(this.foo).query(int)

Having bound tables, specifically, allow to add the useful .query_schema() API:

# CREATE TABLE a (b FLOAT)
>>> schema = {'b': float}
>>> db.table('a', schema=schema).create().query()

# Queries the database for the schema, and returns a new bound table instance
>>> t2 = db.table('a').query_schema()
>>> t2.schema
{'b': Float(precision=5)}

Advanced topics

Dialect Mixins

In Sqeleton, each dialect class represents a specific SQL dialect. Dialects are responsible for providing code fragments to the SQL compiler.

Since Sqeleton aims to support a growing amount of features, and allow custom database implementations, extra features are provided through mixins. That way, when implementing a new database (either in Sqeleton, or in a private code-base), we can pick and choose which features we want to implement, and which ones we don’t. Sqeleton will throw an error if the mixin we’re trying to use isn’t supported by one of the databases we’re using.

The simplest way to load mixins is to use the Connect.load_mixins() methods, and provide the abstract mixins you want to use:

import sqeleton
from sqeleton.abcs.mixins import AbstractMixin_NormalizeValue, AbstractMixin_RandomSample

connect = sqeleton.connect.load_mixins(AbstractMixin_NormalizeValue)
ddb = connect("duckdb://:memory:")
print(ddb.dialect.normalize_boolean("bool", None))
# Outputs:
#   bool::INTEGER::VARCHAR

Each database is already aware of the available mixin implementations, because it was defined with the MIXINS attribute. We’re only using the abstract mixins to select the mixins we want to use.

List of mixins

List of available abstract mixins:

AbstractMixin_NormalizeValue
AbstractMixin_MD5
AbstractMixin_Schema
AbstractMixin_Regex
AbstractMixin_RandomSample
AbstractMixin_TimeTravel - Only snowflake & bigquery
AbstractMixin_OptimizerHints - Only oracle & mysql

More will be added in the future.

Note that it’s still possible to use user-defined mixins that aren’t on this list.

Unimplemented Mixins

Trying to load a mixin that isn’t implemented by all databases, will fail:

>>> from sqeleton.abcs.mixins import AbstractMixin_TimeTravel
>>> connect.load_mixins(AbstractMixin_TimeTravel)
Traceback (most recent call last):
    ...
TypeError: Can't instantiate abstract class PostgresqlDialect with abstract method time_travel

In such a case, it’s possible to use Connect.for_databases() to only load for a subset of the available databases:

# No problem, time travel is implemented in both
# Trying to connect to other databases will fail
connect = sqeleton.connect.for_databases('bigquery', 'snowflake').load_mixins(AbstractMixin_TimeTravel)

The .load_mixins() method is just a convenience method. It’s possible to achieve the same functionality, and with more fine-grained control, using explicit inheritance, and finally creating a new Connect object.

Note that both .load_mixins() and .for_databases() create new instances of Connect, and it’s okay to have more than one at the same time.

Type Inference, mypy, etc.

Python’s typing module doesn’t yet support intersection / multiple-inheritance, and so .load_mixins() can’t provide the necessary information for type-checking.

The recommended solution is to override the type of the database returned from connect():

from sqeleton.abcs import AbstractDialect, AbstractDatabase

class NewAbstractDialect(AbstractDialect, AbstractMixin_NormalizeValue, AbstractMixin_RandomSample):
    pass

connect = sqeleton.connect.load_mixins(AbstractMixin_NormalizeValue, AbstractMixin_RandomSample)
ddb: AbstractDatabase[NewAbstractDialect] = connect("duckdb://:memory:")
# ddb.dialect is now known to implement NewAbstractDialect.

Query interpreter

In addition to query expressions, Database.query() can accept a generator, which will behave as an “interpreter”.

The generator executes queries by yielding them.

Using a query interpreter also guarantees that subsequent calls to .query() will run in the same session. That can be useful for using temporary tables, or session variables.

Example:

def sample_using_temp_table(db: Database, source_table: ITable, sample_size: int):
    "This function creates a temporary table from a query and then samples rows from it"

    results = []

    def _sample_using_temp_table():
        nonlocal results

        yield code("CREATE TEMPORARY TABLE tmp1 AS {source_table}", source_table=source_table)

        tbl = table('tmp1') 
        try:
            results += yield sample(tbl, sample_size)
        finally:
            yield tbl.drop()

    db.query(_sample_using_temp_table())
    return results

Query params

TODO

Other features

SQL client

Sqeleton comes with a simple built-in SQL client, in the form of a REPL, which accepts SQL commands, and a few special commands.

It accepts any database URL that is supported by Sqeleton. That can be useful for querying databases that don’t have established clients.

You can call it using sqeleton repl <url>.

Example:

# Start a REPL session
$ sqeleton repl duckdb:///pii_test.ddb

# Run SQL
DuckDB> select (22::float / 7) as almost_pi
┏━━━━━━━━━━━━━━━━━━━┓
┃ almost_pi         ┃
┡━━━━━━━━━━━━━━━━━━━┩
│ 3.142857074737549 │
└───────────────────┘
       1 rows

# Display help
DuckDB> ?

Commands:
  ?mytable - shows schema of table 'mytable'
  * - shows list of all tables
  *pattern - shows list of all tables with name like pattern
Otherwise, runs regular SQL query

List of supported databases

Database	Status	Connection string
PostgreSQL >=10	💚	`postgresql://<user>:<password>@<host>:5432/<database>`
MySQL	💚	`mysql://<user>:<password>@<hostname>:5432/<database>`
Snowflake	💚	`"snowflake://<user>[:<password>]@<account>/<database>/<SCHEMA>?warehouse=<WAREHOUSE>&role=<role>[&authenticator=externalbrowser]"`
BigQuery	💚	`bigquery://<project>/<dataset>`
Redshift	💚	`redshift://<username>:<password>@<hostname>:5439/<database>`
Oracle	💛	`oracle://<username>:<password>@<hostname>/database`
Presto	💛	`presto://<username>:<password>@<hostname>:8080/<database>`
Databricks	💛	`databricks://<http_path>:<access_token>@<server_hostname>/<catalog>/<schema>`
Trino	💛	`trino://<username>:<password>@<hostname>:8080/<database>`
Clickhouse	💛	`clickhouse://<username>:<password>@<hostname>:9000/<database>`
Vertica	💛	`vertica://<username>:<password>@<hostname>:5433/<database>`
DuckDB	💛
ElasticSearch	📝
Planetscale	📝
Pinot	📝
Druid	📝
Kafka	📝
SQLite	📝

💚: Implemented and thoroughly tested.
💛: Implemented, but not thoroughly tested yet.
⏳: Implementation in progress.
📝: Implementation planned. Contributions welcome.

Is your database not listed here? We accept pull-requests!

Connection Editor

A common complaint among new users was the difficulty in setting up the connections.

Connection URLs are admittedly confusing, and editing .toml files isn’t always straight-forward either.

To ease this initial difficulty, we added a textual-based TUI tool to sqeleton, that allows users to edit configuration files and test the connections while editing them.

Install

This tool needs textual to run. You can install it using:

pip install 'sqeleton[tui]'

Make sure you also have drivers for whatever database connection you’re going to edit!

Run

Once everything is installed, you can run the editor with the following command:

sqeleton conn-editor <conf_path.toml>

Example:

sqeleton conn-editor ~/dbs.toml

The available actions and hotkeys will be listed in the status bar.

Note: using the connection editor will delete comments and reformat the file!

We recommend backing up the configuration file before editing it.

Python API Reference

User API

Database

exception sqeleton.databases.base.ConnectError

exception sqeleton.databases.base.QueryError

class sqeleton.databases.base.ThreadLocalInterpreter(compiler: Compiler, gen: Generator)

An interpeter used to execute a sequence of queries within the same thread and cursor.

Useful for cursor-sensitive operations, such as creating a temporary table.

class sqeleton.databases.base.Mixin_Schema

table_information() → Compilable: Query to return a table of schema information about existing tables

list_tables(table_schema: str, like: Optional[Compilable] = None) → Compilable

Query to select the list of tables in the schema. (query return type: table[str])

If ‘like’ is specified, the value is applied to the table name, using the ‘like’ operator.

class sqeleton.databases.base.Mixin_RandomSample

random_sample_n(tbl: AbstractTable, size: int) → AbstractTable: Take a random sample of the given size, i.e. return ‘size’ amount of rows

random_sample_ratio_approx(tbl: AbstractTable, ratio: float) → AbstractTable

Take a random sample of the approximate size determined by the ratio (0..1), where 0 means no rows, and 1 means all rows

i.e. the actual mount of rows returned may vary by standard deviation.

class sqeleton.databases.base.Mixin_OptimizerHints

optimizer_hints(hints: str) → str

Creates a compatible optimizer_hints string

Parameters:: hints (optimizer_hints - string of optimizer) –

class sqeleton.databases.base.BaseDialect

offset_limit(offset: Optional[int] = None, limit: Optional[int] = None): Provide SQL fragment for limit and offset inside a select

concat(items: List[str]) → str: Provide SQL for concatenating a bunch of columns into a string

is_distinct_from(a: str, b: str) → str: Provide SQL for a comparison where NULL = NULL is true

timestamp_value(t: datetime) → str: Provide SQL for the given timestamp value

random() → str: Provide SQL for generating a random number betweein 0..1

current_timestamp() → str: Provide SQL for returning the current timestamp, aka now

explain_as_text(query: str) → str: Provide SQL for explaining a query, returned as table(varchar)

classmethod load_mixins(*abstract_mixins) → Any: Load a list of mixins that implement the given abstract mixins

class sqeleton.databases.base.QueryResult(rows: list[Any] = <object object at 0x7fed69786ee0>, columns: (NoneType+list[Any]) = None)

class sqeleton.databases.base.Database(*args, **kwds)

Base abstract class for databases.

Used for providing connection code and implementation specific SQL utilities.

Instanciated using connect()

query(sql_ast: Union[ExprNode, str, bool, int, float, datetime, ArithString, None, Generator], res_type: Optional[type] = None)

Query the given SQL code/AST, and attempt to convert the result to type ‘res_type’

If given a generator, it will execute all the yielded sql queries with the same thread and cursor. The results of the queries a returned by the yield stmt (using the .send() mechanism). It’s a cleaner approach than exposing cursors, but may not be enough in all cases.

select_table_schema(path: Tuple[str, ...]) → str: Provide SQL for selecting the table schema as (name, type, date_prec, num_prec)

query_table_schema(path: Tuple[str, ...]) → Dict[str, tuple]

Query the table for its schema for table in ‘path’, and return {column: tuple} where the tuple is (table_name, col_name, type_repr, datetime_precision?, numeric_precision?, numeric_scale?)

Note: This method exists instead of select_table_schema(), just because not all databases support: accessing the schema using a SQL query.

select_table_unique_columns(path: Tuple[str, ...]) → str: Provide SQL for selecting the names of unique columns in the table

query_table_unique_columns(path: Tuple[str, ...]) → List[str]: Query the table for its unique columns for table in ‘path’, and return {column}

parse_table_name(name: str) → Tuple[str, ...]: Parse the given table name into a DbPath

close(): Close connection(s) to the database instance. Querying will stop functioning.

classmethod load_mixins(*abstract_mixins) → type: Extend the dialect with a list of mixins that implement the given abstract mixins.

class sqeleton.databases.base.ThreadedDatabase(thread_count=1)

Access the database through singleton threads.

Used for database connectors that do not support sharing their connection between different threads.

abstract create_connection(): Return a connection instance, that supports the .cursor() method.

close(): Close connection(s) to the database instance. Querying will stop functioning.

property is_autocommit: bool: Return whether the database autocommits changes. When false, COMMIT statements are skipped.

Queries

sqeleton.queries.api.join(*tables: ITable) → Join

Inner-join a sequence of table expressions”

When joining, it’s recommended to use explicit tables names, instead of this, in order to avoid potential name collisions.

Example

person = table('person')
city = table('city')

name_and_city = (
    join(person, city)
    .on(person['city_id'] == city['id'])
    .select(person['id'], city['name'])
)

sqeleton.queries.api.leftjoin(*tables: ITable)

Left-joins a sequence of table expressions.

Internals

This section is for developers who wish to improve sqeleton, or to extend it within their own project.

Regular users might also find it useful for debugging and understanding, especially at this early stage of the project.

Query ASTs

exception sqeleton.queries.ast_classes.QueryBuilderError

exception sqeleton.queries.ast_classes.QB_TypeError

class sqeleton.queries.ast_classes.ExprNode: Base class for query expression nodes

class sqeleton.queries.ast_classes.Code(code: str = <object object at 0x7fed69786ee0>, args: (NoneType+dict[(str*(NoneType+ExprNode+bool+str+int+float+ArithString+datetime))]) = None)

class sqeleton.queries.ast_classes.Alias(expr: (NoneType+ExprNode+bool+str+int+float+ArithString+datetime) = <object object at 0x7fed69786ee0>, name: str = <object object at 0x7fed69786ee0>)

class sqeleton.queries.ast_classes.ITable

select(*exprs, distinct=SKIP, optimizer_hints=SKIP, **named_exprs) → ITable: Create a new table with the specified fields

where(*exprs): Filter the rows, based on the given predicates. (aka Selection)

order_by(*exprs): Order the rows lexicographically, according to the given expressions.

limit(limit: int): Stop yielding rows after the given limit. i.e. take the first ‘n=limit’ rows

join(target: ITable): Join this table with the target table.

group_by(*keys) → GroupBy

Group according to the given keys.

Must be followed by a call to :ref:GroupBy.agg()

count(): SELECT count() FROM self

union(other: ITable): SELECT * FROM self UNION other

union_all(other: ITable): SELECT * FROM self UNION ALL other

minus(other: ITable): SELECT * FROM self EXCEPT other

intersect(other: ITable): SELECT * FROM self INTERSECT other

class sqeleton.queries.ast_classes.Concat(exprs: list[Any] = <object object at 0x7fed69786ee0>, sep: (NoneType+str) = None)

class sqeleton.queries.ast_classes.Count(expr: NoneType + NoneType + ExprNode + bool + str + int + float + ArithString + datetime = None, distinct: bool = False)

type: alias of int

class sqeleton.queries.ast_classes.TestRegex(string: (NoneType+ExprNode+bool+str+int+float+ArithString+datetime) = <object object at 0x7fed69786ee0>, pattern: (NoneType+ExprNode+bool+str+int+float+ArithString+datetime) = <object object at 0x7fed69786ee0>)

class sqeleton.queries.ast_classes.Func(name: str = <object object at 0x7fed69786ee0>, args: Sequence[(NoneType+ExprNode+bool+str+int+float+ArithString+datetime)] = <object object at 0x7fed69786ee0>)

class sqeleton.queries.ast_classes.WhenThen(when: (NoneType+ExprNode+bool+str+int+float+ArithString+datetime) = <object object at 0x7fed69786ee0>, then: (NoneType+ExprNode+bool+str+int+float+ArithString+datetime) = <object object at 0x7fed69786ee0>)

class sqeleton.queries.ast_classes.CaseWhen(cases: Sequence[WhenThen] = <object object at 0x7fed69786ee0>, else_expr: (NoneType+(NoneType+ExprNode+bool+str+int+float+ArithString+datetime)) = None)

when(*whens: Optional[Union[ExprNode, str, bool, int, float, datetime, ArithString]]) → QB_When

Add a new ‘when’ clause to the case expression

Must be followed by a call to .then()

else_(then: Optional[Union[ExprNode, str, bool, int, float, datetime, ArithString]])

Add an ‘else’ clause to the case expression.

Can only be called once!

class sqeleton.queries.ast_classes.QB_When(casewhen: CaseWhen = <object object>, when: (NoneType+ExprNode+bool+str+int+float+ArithString+datetime) = <object object>)

Partial case-when, used for query-building

then(then: Optional[Union[ExprNode, str, bool, int, float, datetime, ArithString]]) → CaseWhen: Add a ‘then’ clause after a ‘when’ was added.

class sqeleton.queries.ast_classes.IsDistinctFrom(a: (NoneType+ExprNode+bool+str+int+float+ArithString+datetime) = <object object at 0x7fed69786ee0>, b: (NoneType+ExprNode+bool+str+int+float+ArithString+datetime) = <object object at 0x7fed69786ee0>)

type: alias of bool

class sqeleton.queries.ast_classes.BinOp(op: str = <object object at 0x7fed69786ee0>, args: Sequence[(NoneType+ExprNode+bool+str+int+float+ArithString+datetime)] = <object object at 0x7fed69786ee0>)

class sqeleton.queries.ast_classes.UnaryOp(op: str = <object object at 0x7fed69786ee0>, expr: (NoneType+ExprNode+bool+str+int+float+ArithString+datetime) = <object object at 0x7fed69786ee0>)

class sqeleton.queries.ast_classes.BinBoolOp(op: str = <object object>, args: Sequence[(NoneType+ExprNode+bool+str+int+float+ArithString+datetime)] = <object object>)

type: alias of bool

class sqeleton.queries.ast_classes.Column(source_table: ITable = <object object at 0x7fed69786ee0>, name: str = <object object at 0x7fed69786ee0>)

class sqeleton.queries.ast_classes.TablePath(path: tuple[str] = <object object at 0x7fed69786ee0>, schema: (CaseAwareMapping+NoneType) = None)

create(source_table: Optional[ITable] = None, *, if_not_exists: bool = False, primary_keys: Optional[List[str]] = None)

Returns a query expression to create a new table.

Parameters:

source_table – a table expression to use for initializing the table. If not provided, the table must have a schema specified.
if_not_exists – Add a ‘if not exists’ clause or not. (note: not all dbs support it!)
primary_keys – List of column names which define the primary key

drop(if_exists=False)

Returns a query expression to delete the table.

Parameters:: if_not_exists – Add a ‘if not exists’ clause or not. (note: not all dbs support it!)

truncate(): Returns a query expression to truncate the table. (remove all rows)

insert_rows(rows: Sequence, *, columns: Optional[List[str]] = None)

Returns a query expression to insert rows to the table, given as Python values.

Parameters:

rows – A list of tuples. Must all have the same width.
columns – Names of columns being populated. If specified, must have the same length as the tuples.

insert_row(*values, columns: Optional[List[str]] = None)

Returns a query expression to insert a single row to the table, given as Python values.

Parameters:: columns – Names of columns being populated. If specified, must have the same length as ‘values’

insert_expr(expr: Optional[Union[ExprNode, str, bool, int, float, datetime, ArithString]])

Returns a query expression to insert rows to the table, given as a query expression.

Parameters:: expr – query expression to from which to read the rows

time_travel(*, before: bool = False, timestamp: Optional[datetime] = None, offset: Optional[int] = None, statement: Optional[str] = None) → Compilable

Selects historical data from the table

Parameters:

before – If false, inclusive of the specified point in time. If True, only return the time before it. (at/before)
timestamp – A constant timestamp
offset – the time ‘offset’ seconds before now
statement – identifier for statement, e.g. query ID

Must specify exactly one of timestamp, offset or statement.

class sqeleton.queries.ast_classes.TableAlias(source_table: ITable = <object object at 0x7fed69786ee0>, name: str = <object object at 0x7fed69786ee0>)

class sqeleton.queries.ast_classes.Join(source_tables: Sequence[ITable] = <object object at 0x7fed69786ee0>, op: (NoneType+str) = None, on_exprs: (Sequence[(NoneType+ExprNode+bool+str+int+float+ArithString+datetime)]+NoneType) = None, columns: (Sequence[(NoneType+ExprNode+bool+str+int+float+ArithString+datetime)]+NoneType) = None)

on(*exprs) → Join: Add an ON clause, for filtering the result of the cartesian product (i.e. the JOIN)

select(*exprs, **named_exprs) → ITable

Select fields to return from the JOIN operation

Query Compiler

exception sqeleton.queries.compiler.CompileError

class sqeleton.queries.compiler.Root: Nodes inheriting from Root can be used as root statements in SQL (e.g. SELECT yes, RANDOM() no)

class sqeleton.queries.compiler.Compiler(database: AbstractDatabase = <object object at 0x7fed69786ee0>, params: dict[(Any*Any)] = <factory>, in_select: bool = False, in_join: bool = False, _table_context: list[Any] = <factory>, _subqueries: dict[(str*Any)] = <factory>, root: bool = True, _counter: list[Any] = <factory>)

ABCS

class sqeleton.abcs.database_types.PrecisionType(precision: int = <object object at 0x7fed69786ee0>, rounds: (Unknown+bool) = Unknown)

class sqeleton.abcs.database_types.TemporalType(precision: int = <object object>, rounds: (Unknown+bool) = Unknown)

class sqeleton.abcs.database_types.Timestamp(precision: int = <object object>, rounds: (Unknown+bool) = Unknown)

class sqeleton.abcs.database_types.TimestampTZ(precision: int = <object object>, rounds: (Unknown+bool) = Unknown)

class sqeleton.abcs.database_types.Datetime(precision: int = <object object>, rounds: (Unknown+bool) = Unknown)

class sqeleton.abcs.database_types.Date(precision: int = <object object>, rounds: (Unknown+bool) = Unknown)

class sqeleton.abcs.database_types.NumericType(precision: int = <object object at 0x7fed69786ee0>)

class sqeleton.abcs.database_types.FractionalType(precision: int = <object object>)

class sqeleton.abcs.database_types.Float(precision: int = <object object>)

python_type: alias of float

class sqeleton.abcs.database_types.IKey

Interface for ColType, for using a column as a key in table.

abstract property python_type: type: Return the equivalent Python type of the key

class sqeleton.abcs.database_types.Decimal(precision: int = <object object>)

property python_type: type: Return the equivalent Python type of the key

class sqeleton.abcs.database_types.StringType

python_type: alias of str

class sqeleton.abcs.database_types.ColType_UUID

python_type: alias of ArithUUID

class sqeleton.abcs.database_types.ColType_Alphanum

python_type: alias of ArithAlphanumeric

class sqeleton.abcs.database_types.Native_UUID

class sqeleton.abcs.database_types.String_UUID

class sqeleton.abcs.database_types.String_Alphanum

class sqeleton.abcs.database_types.String_VaryingAlphanum

class sqeleton.abcs.database_types.String_FixedAlphanum(length: int = <object object at 0x7fed69786ee0>)

class sqeleton.abcs.database_types.Text

class sqeleton.abcs.database_types.Integer(precision: int = 0, python_type: type = <class 'int'>)

python_type: alias of int

class sqeleton.abcs.database_types.UnknownColType(text: str = <object object at 0x7fed69786ee0>)

class sqeleton.abcs.database_types.AbstractDialect

Dialect-dependent query expressions

abstract property name: str: Name of the dialect

abstract classmethod load_mixins(*abstract_mixins) → Any: Load a list of mixins that implement the given abstract mixins

abstract property ROUNDS_ON_PREC_LOSS: bool: True if db rounds real values when losing precision, False if it truncates.

abstract quote(s: str): Quote SQL name

abstract concat(items: List[str]) → str: Provide SQL for concatenating a bunch of columns into a string

abstract is_distinct_from(a: str, b: str) → str: Provide SQL for a comparison where NULL = NULL is true

abstract to_string(s: str) → str: Provide SQL for casting a column to string

abstract random() → str: Provide SQL for generating a random number betweein 0..1

abstract current_timestamp() → str: Provide SQL for returning the current timestamp, aka now

abstract offset_limit(offset: Optional[int] = None, limit: Optional[int] = None): Provide SQL fragment for limit and offset inside a select

abstract explain_as_text(query: str) → str: Provide SQL for explaining a query, returned as table(varchar)

abstract timestamp_value(t: datetime) → str: Provide SQL for the given timestamp value

abstract set_timezone_to_utc() → str: Provide SQL for setting the session timezone to UTC

abstract parse_type(table_path: Tuple[str, ...], col_name: str, type_repr: str, datetime_precision: Optional[int] = None, numeric_precision: Optional[int] = None, numeric_scale: Optional[int] = None) → ColType: Parse type info as returned by the database

class sqeleton.abcs.database_types.AbstractDatabase(*args, **kwds)

abstract property dialect: T_Dialect: The dialect of the database. Used internally by Database, and also available publicly.

abstract classmethod load_mixins(*abstract_mixins) → type: Extend the dialect with a list of mixins that implement the given abstract mixins.

abstract property CONNECT_URI_HELP: str: Example URI to show the user in help and error messages

abstract property CONNECT_URI_PARAMS: List[str]: List of parameters given in the path of the URI

abstract query_table_schema(path: Tuple[str, ...]) → Dict[str, tuple]

Query the table for its schema for table in ‘path’, and return {column: tuple} where the tuple is (table_name, col_name, type_repr, datetime_precision?, numeric_precision?, numeric_scale?)

Note: This method exists instead of select_table_schema(), just because not all databases support: accessing the schema using a SQL query.

abstract select_table_unique_columns(path: Tuple[str, ...]) → str: Provide SQL for selecting the names of unique columns in the table

abstract query_table_unique_columns(path: Tuple[str, ...]) → List[str]: Query the table for its unique columns for table in ‘path’, and return {column}

abstract parse_table_name(name: str) → Tuple[str, ...]: Parse the given table name into a DbPath

abstract close(): Close connection(s) to the database instance. Querying will stop functioning.

abstract property is_autocommit: bool: Return whether the database autocommits changes. When false, COMMIT statements are skipped.

class sqeleton.abcs.database_types.AbstractTable

abstract select(*exprs, distinct=False, **named_exprs) → AbstractTable

Choose new columns, based on the old ones. (aka Projection)

Parameters:

exprs – List of expressions to constitute the columns of the new table. If not provided, returns all columns in source table (i.e. select *)
distinct – ‘select’ or ‘select distinct’
named_exprs – More expressions to constitute the columns of the new table, aliased to keyword name.

abstract where(*exprs) → AbstractTable: Filter the rows, based on the given predicates. (aka Selection)

abstract order_by(*exprs) → AbstractTable: Order the rows lexicographically, according to the given expressions.

abstract limit(limit: int) → AbstractTable: Stop yielding rows after the given limit. i.e. take the first ‘n=limit’ rows

abstract join(target) → AbstractTable

Join the current table with the target table, returning a new table containing both side-by-side.

When joining, it’s recommended to use explicit tables names, instead of this, in order to avoid potential name collisions.

Example

person = table('person')
city = table('city')

name_and_city = (
    person
    .join(city)
    .on(person['city_id'] == city['id'])
    .select(person['id'], city['name'])
)

abstract group_by(*keys)

Behaves like in SQL, except for a small change in syntax:

A call to .agg() must follow every call to .group_by().

Example

# SELECT a, sum(b) FROM tmp GROUP BY 1
table('tmp').group_by(this.a).agg(this.b.sum())

# SELECT a, sum(b) FROM a GROUP BY 1 HAVING (b > 10)
(table('tmp')
    .group_by(this.a)
    .agg(this.b.sum())
    .having(this.b > 10)
)

abstract count() → int: SELECT count() FROM self

abstract union(other: ITable): SELECT * FROM self UNION other

abstract union_all(other: ITable): SELECT * FROM self UNION ALL other

abstract minus(other: ITable): SELECT * FROM self EXCEPT other

abstract intersect(other: ITable): SELECT * FROM self INTERSECT other

class sqeleton.abcs.mixins.AbstractMixin: A mixin for a database dialect

class sqeleton.abcs.mixins.AbstractMixin_NormalizeValue

abstract normalize_timestamp(value: str, coltype: TemporalType) → str

Creates an SQL expression, that converts ‘value’ to a normalized timestamp.

The returned expression must accept any SQL datetime/timestamp, and return a string.

Date format: YYYY-MM-DD HH:mm:SS.FFFFFF

Precision of dates should be rounded up/down according to coltype.rounds

abstract normalize_number(value: str, coltype: FractionalType) → str

Creates an SQL expression, that converts ‘value’ to a normalized number.

The returned expression must accept any SQL int/numeric/float, and return a string.

Floats/Decimals are expected in the format “I.P”

Where I is the integer part of the number (as many digits as necessary), and must be at least one digit (0). P is the fractional digits, the amount of which is specified with coltype.precision. Trailing zeroes may be necessary. If P is 0, the dot is omitted.

Note: We use ‘precision’ differently than most databases. For decimals, it’s the same as numeric_scale, and for floats, who use binary precision, it can be calculated as log10(2**numeric_precision).

normalize_boolean(value: str, _coltype: Boolean) → str: Creates an SQL expression, that converts ‘value’ to either ‘0’ or ‘1’.

normalize_uuid(value: str, coltype: ColType_UUID) → str: Creates an SQL expression, that strips uuids of artifacts like whitespace.

normalize_value_by_type(value: str, coltype: ColType) → str

Creates an SQL expression, that converts ‘value’ to a normalized representation.

The returned expression must accept any SQL value, and return a string.

The default implementation dispatches to a method according to coltype:

TemporalType    -> normalize_timestamp()
FractionalType  -> normalize_number()
*else*          -> to_string()

(`Integer` falls in the *else* category)

class sqeleton.abcs.mixins.AbstractMixin_MD5

Methods for calculating an MD6 hash as an integer.

abstract md5_as_int(s: str) → str: Provide SQL for computing md5 and returning an int

class sqeleton.abcs.mixins.AbstractMixin_Schema

Methods for querying the database schema

TODO: Move AbstractDatabase.query_table_schema() and friends over here

table_information() → Compilable: Query to return a table of schema information about existing tables

abstract list_tables(table_schema: str, like: Optional[Compilable] = None) → Compilable

Query to select the list of tables in the schema. (query return type: table[str])

If ‘like’ is specified, the value is applied to the table name, using the ‘like’ operator.

class sqeleton.abcs.mixins.AbstractMixin_Regex

abstract test_regex(string: Compilable, pattern: Compilable) → Compilable: Tests whether the regex pattern matches the string. Returns a bool expression.

class sqeleton.abcs.mixins.AbstractMixin_RandomSample

abstract random_sample_n(tbl: str, size: int) → str: Take a random sample of the given size, i.e. return ‘size’ amount of rows

abstract random_sample_ratio_approx(tbl: str, ratio: float) → str

Take a random sample of the approximate size determined by the ratio (0..1), where 0 means no rows, and 1 means all rows

i.e. the actual mount of rows returned may vary by standard deviation.

class sqeleton.abcs.mixins.AbstractMixin_TimeTravel

abstract time_travel(table: Compilable, before: bool = False, timestamp: Optional[Compilable] = None, offset: Optional[Compilable] = None, statement: Optional[Compilable] = None) → Compilable

Selects historical data from a table

Parameters:

querying (table - The name of the table whose history we're) –
timestamp (timestamp - A constant) –
now (offset - the time 'offset' seconds before) –
statement (statement - identifier for) –
ID (e.g. query) –

Must specify exactly one of timestamp, offset or statement.

class sqeleton.abcs.mixins.AbstractMixin_OptimizerHints

abstract optimizer_hints(optimizer_hints: str) → str

Creates a compatible optimizer_hints string

Parameters:: hints (optimizer_hints - string of optimizer) –

Sqeleton

Sqeleton is a Python library for querying SQL databases.

It consists of -

A fast and concise query builder, inspired by PyPika and SQLAlchemy
A modular database interface, with drivers for a long list of SQL databases.

It is comparable to other libraries such as SQLAlchemy or PyPika, in terms of API and intended audience. However there are several notable ways in which it is different.

For more information, See our README

Resources

Install / Get started
Introduction to Sqeleton
List of supported databases
Connection Editor
Python API Reference
Source code (git): https://github.com/datafold/sqeleton

Install / Get started

Database drivers

Connection editor

What’s next?

Introduction to Sqeleton

Database Interface

connect()

Database.query()

Query Builder

table()

table attributes and this

.select(), .where(), .order_by(), .limit()

.group_by(), .agg(), and .having()

More table operations

.join(), .on()

when() and .then()

DDL - .create(), .drop(), .truncate()

Add data - .insert_row(), .insert_rows(), .insert_expr()

Raw SQL using code()

Bound Tables & Expressions - Database.table()

Advanced topics

Dialect Mixins

List of mixins

Unimplemented Mixins

Type Inference, mypy, etc.

Query interpreter

Query params

Other features

SQL client

List of supported databases

Connection Editor

Install

Run

Python API Reference

User API

Database

Queries

Internals

Query ASTs

Query Compiler

ABCS

Sqeleton

Resources

table attributes and `this`

`when()` and `.then()`

Raw SQL using `code()`