SQL Query Builder
The goal of the SQL Query Builder is to provide a canonical interface to build SQL queries. The QueryBuilder refers to a dialect as a variant of the SQL language dependent of the choice of the backend technology.
The only supported dialect is the one of SQLite but building new dialects shouldn’t be too hard.
The submodule sql contains many standard constructors for SQL languages. See the submodule sqlite for a dialect for SQLite.
Typical usage
Building a Schema
We first import the generic SQL dialect.
>>> from networkdisk.sql import sqldialect as dialect
The SQL dialect contains pointers towards several subdialects:
>>> dialect
Dialect(SQL)<constraints, conditions, constants, columns, queries, tupledict, schemata, graph_schema, master, schema, helper, Graph, DiGraph>
Each subdialect contains an SQL query builder object.
>>> c = dialect.columns.ValueColumn(1)
>>> c2 = dialect.columns.AddColumn(c, c)
>>> dialect.queries.SelectQuery(columns=c2)
SelectQuery<SELECT ?+?>
We can describe an abstract Relational Schema.
>>> Schema = dialect.schema.Schema()
>>> fooTable = Schema.add_table("foo")
>>> idC = fooTable.add_column("id", primarykey=True, sqltype="INT")
>>> nameC = fooTable.add_column("name", sqltype="TEXT")
>>> ageC = fooTable.add_column("age", sqltype="INT")
>>> fooTable
SchemaTable<foo(id, name, age)>
The object fooTable is simply an abstract representation of the schema. Remark that the order of the columns matters. This order is accessible:
>>> fooTable[0]
SchemaTableColumn<id>
>>> fooTable[1]
SchemaTableColumn<name>
>>> fooTable[2]
SchemaTableColumn<age>
It is also possible to resolve columns:
>>> fooTable.get("age")
SchemaTableColumn<age>
>>> fooTable["age"]
SchemaTableColumn<age>
This table signature can be created as follows:
>>> fooTable.create_query()
CreateTableQuery<CREATE TABLE foo (id INT PRIMARY KEY, name TEXT, age INT)>
It is possible to declare an index on the table.
>>> Schema.add_index(fooTable, (nameC, ageC))
SchemaIndex<CREATE INDEX index_foo_name_age ON foo(name, age)>
Remark that it can use column resolution here as well:
>>> Schema.add_index(fooTable, ("id", "name"))
SchemaIndex<CREATE INDEX index_foo_id_name ON foo(id, name)>
We can easily create another table, with references and foreign-key constaint:
>>> barTable = Schema.add_table("bar")
>>> fkey_cstrt = dialect.constraints.ForeignkeyConstraint(idC, container=idC.container_name, on_delete="CASCADE")
>>> fidC = barTable.add_column("foreign_id", references=idC, constraints=(fkey_cstrt,)) # no need to gives the type
>>> msgC = barTable.add_column("msg", sqltype="TEXT")
The resulting object is a Query that can be executed on the database. The actual query as string can be simply obtained using str and directly feed to a db-connector.
>>> str(fooTable.create_query())
'CREATE TABLE foo (id INT PRIMARY KEY, name TEXT, age INT)'
Finally, we can create the whole schema, (the table and its indices) with the create_script method of Schema
>>> print(",\n".join(map(str, Schema.create_script()))) # it returns a generator
CREATE TABLE foo (id INT PRIMARY KEY, name TEXT, age INT),
CREATE INDEX index_foo_name_age ON foo(name, age),
CREATE INDEX index_foo_id_name ON foo(id, name),
CREATE TABLE bar (foreign_id INT REFERENCES foo(id) ON DELETE CASCADE, msg TEXT)
Querying a schema
Once a SQL schema is defined, we can interact with it relatively easily.
It is possible construct some queries easily.
>>> fooTable.select_query()
SelectQuery<SELECT id, name, age FROM foo>
>>> fooTable.select_query(columns=("age",))
SelectQuery<SELECT age FROM foo>
We can add a WHERE clause by generating condition:
>>> fooTable.select_query(condition=ageC.eq(18))
SelectQuery<SELECT id, name, age FROM foo WHERE age = ?>
>>> fooTable.select_query(condition=ageC.gt(18) & ageC.lt(42))
SelectQuery<SELECT id, name, age FROM foo WHERE age > ? AND age < ?>
Here, constant are provided as argument in the connector, they thus appear as a question mark ? placeholder.
Simple join can be perform as well:
>>> JQ= fooTable.inner_join_query(barTable, (idC, fidC))
>>> JQ
JoinQuery<foo INNER JOIN bar ON id = foreign_id>
A JoinQuery is not executable directly:
>>> JQ.select_query(columns=("name", "msg"))
SelectQuery<SELECT name, msg FROM foo INNER JOIN bar ON id = foreign_id>
>>> JQ.select_query(columns=("name", "msg"), condition=msgC.neq(dialect.constants.Null))
SelectQuery<SELECT name, msg FROM foo INNER JOIN bar ON id = foreign_id WHERE msg != ?>
We can define a query that insert some data into the table.
>>> Insert = fooTable.insert_values((0, "John Smith", 42), (1, "Jane Doe", 43))
>>> Insert
InsertQuery<INSERT INTO foo(id, name, age) VALUES (?, ?, ?), (?, ?, ?)>
The resulting object take cares of the encoding of the value. It takes the shape of a data structure mapping each positional ? to an encoder with the appropriate position. The encoder can be customized depending on the provided SQLType.
Dedicated queries can be used to perform bulk insertion and avoid multiplying queries. For more information on querying with an SQL dialect, see the Query reference page.
The SQLite Dialect and real DB manipulation
Unlike the SQL Dialect, the SQLite Dialect can be actually used to manipulate an SQLite database. It provides an actual helper class that provides a common interface to the DB-connector.
>>> from networkdisk.sqlite.dialect import sqlitedialect as dialect # import the dialect
>>> helper = dialect.helper.Helper(":memory:") # in-memory SQLite database.
>>> helper
Helper〈dialect=SQLite, dbpath=:memory: 〉
We can turn SQL Logging to see the actual query send to the DB:
>>> helper.sql_logger.active = True
>>> helper.sql_logger.color = False # Turn off coloring for rendering
The previous SQL code is valid for SQLite as is (but have to be generated using the sqlite dialect):
>>> Schema = dialect.schema.Schema()
>>> fooTable = Schema.add_table("foo")
>>> idC = fooTable.add_column("id", primarykey=True, sqltype="INT")
>>> nameC = fooTable.add_column("name", sqltype="TEXT")
>>> ageC = fooTable.add_column("age", sqltype="INT")
>>> Idx1 = Schema.add_index(fooTable, ("id", "name"))
>>> Idx2 = Schema.add_index(fooTable, (nameC, ageC))
>>> barTable = Schema.add_table("bar")
>>> fkey_cstrt = dialect.constraints.ForeignkeyConstraint(idC, container=idC.container_name, on_delete="CASCADE")
>>> fidC = barTable.add_column("foreign_id", references=idC, constraints=(fkey_cstrt,))
>>> msgC = barTable.add_column("msg", sqltype="TEXT")
We can then create the script:
>>> cursor = helper.executescript(Schema.create_script())
show/hide
CREATE TABLE foo (id INT PRIMARY KEY, name TEXT, age INT)
CREATE INDEX index_foo_id_name ON foo(id, name)
CREATE INDEX index_foo_name_age ON foo(name, age)
CREATE TABLE bar (foreign_id INT REFERENCES foo(id) ON DELETE CASCADE, msg TEXT)
We can turn down the logger:
>>> helper.sql_logger.active = False
And try to insert some values:
>>> cursor = helper.execute(fooTable.insert_values((0, "John Smith", 42), (1, "Jane Doe", 43)))
And select some data:
>>> list(helper.execute(fooTable.select_query(columns=("name",), condition=fooTable["age"].gt(42))))
[('Jane Doe',)]