Let you build SQL requests in pythonic way. SQL blocks could be nested and flexible as you want it to be, constructed dynamically and look pretty! You could also build request once and cache it (but still use variables to replace dynamically). Solid solution!
sqlconstructor is simple, yet very flexible, sql building tool.
You could install from PyPi:
$ python3 -m pip install sqlconstructor
from sqlconstructor import SqlQuery, SqlContainer
# create SqlQuery instance
q = SqlQuery()
# register as many SqlSection instances as you'd like
q['select'](
'id',
'name',
)
q['from'](
'product'
)
q['where'](
"quality = 'Best'",
'and brand_id = 1',
)
# build query into SqlContainer
container: SqlContainer = q()
# get sql as string
sql_text: str = str(container)
SqlSection automatically transforms all sql keywords in uppercase. It does not upper in following cases:
SqlSection automatically adds comma between provided sql statements in __call__ method for next headers:
Output of sql_text is:
SELECT
id,
name
FROM
product
WHERE
quality = 'Best'
AND brand_id = 1
In release >= 1.4.0 it is possible to add default settings for all future sections by special kwargs in SqlQuery constructor such as header_end, sep, line_end, section_end, body_ind, header_ind, do_upper_keywords.
from sqlconstructor import SqlQuery
q = SqlQuery(do_upper_keywords=False)
It is possible to give special keyword arguments in SqlSection __call__ method:
...
q['where'](
"quality = 'Best'",
'brand_id = 1',
header_ind=2,
body_ind=4,
# delimeter = sep + line_end
sep=' AND',
line_end=' ',
section_end=';',
do_upper_keywords=False,
)
...
from sqlconstructor import SqlQuery
q = SqlQuery(
{
'select': (
'id',
'name',
),
'from': 'product',
'where': (
"quality = 'Best'",
'and brand_id = 1',
),
}
)
But it has certain limitation:
But it is possible:
Example of using SqlSectionHeader:
from sqlconstructor import SqlQuery, SqlVal, SqlSectionHeader
h = SqlSectionHeader
q = SqlQuery(
{
h('select'): "'hello'",
# it is possible to add sql section without body
h('union all'): None,
h('select'): SqlVal('hello'),
# or create sql section without header
h(): 'union all',
h('select'): "'hello'",
}
)
container: SqlContainer = q()
sql_text: str = str(container)
Output of sql_text is:
SELECT
'hello'
UNION ALL
SELECT
'hello'
UNION ALL
SELECT
'hello'
from sqlconstructor import SqlQuery
q = SqlQuery()
q['select'](
'id',
'name',
)
q['from'](
'product'
)
q['where'](
"quality = 'Best'",
'and brand_id = 1',
)
for section in q:
# add empty line for "from" statement
if section.header == 'from':
# change container's text in place
section.container.text += '\n'
...
It is possible to slice query (in release >= 1.2.10). When you slice you get new SqlQuery instance. But it is not deep copied (you get new query but with same SqlSection instance elements).
It is possible to get certain SqlSection instance by getitem with integer argument (in release >= 1.2.11). SqlQuery has list behavior (except __init__, __iadd__ and __getitem__ methods) in release >= 1.3.0.
from sqlconstructor import SqlQuery
q = SqlQuery()
q['select'](
'id',
'name',
)
q['from'](
'product'
)
q['where'](
"quality = 'Best'",
'and brand_id = 1',
)
# get only SELECT and FROM statements
new_query = q[:2]
# get first SqlSection
select = q[0]
...
It is possible to append string or any SqlContainer to query as new SqlSection without header in this way:
from sqlconstructor import SqlQuery
def main():
q = SqlQuery()
q += '-- some comment here'
with open('./some_file.sql', encoding='utf-8') as file:
sql_file = file.read().rstrip()
q += sql_file
q['select'](
'p.id',
'p.name',
)
It is possible add extra indentation (positive or negative) for string or SqlContainer by 'add' method (in release >= 1.2.5)
from sqlconstructor import SqlQuery
def main():
q = SqlQuery()
q += '-- some comment here'
with open('./some_file.sql', encoding='utf-8') as file:
sql_file = file.read().rstrip()
q.add(sql_file, ind=4)
...
from sqlconstructor import SqlQuery, SqlContainer
def main():
q = SqlQuery()
q['select'](
'p.id',
'p.name',
)
q += get_from_statement()
...
def get_from_statement() -> SqlContainer:
...
In release >= 1.2.4 it is possible to nest SqlQuery instance (it will be converted to SqlContainer automatically).
from sqlconstructor import SqlQuery, SqlContainer
def main():
q = SqlQuery()
q['select'](
'p.id',
'p.name',
)
q += get_from_statement()
q += get_where_filters()
...
def get_from_statement() -> SqlQuery:
...
def get_where_filters() -> SqlQuery:
...
In release >= 1.3.13
from sqlconstructor import SqlQuery
q = SqlQuery()
q.add(
{
'select': (
'product_id',
'price'
),
'from': 'prices',
}
)
Add section settings in dict (in release >= 1.4.0)
from sqlconstructor import SqlQuery
q = SqlQuery()
q.add(
{
'select': (
'a',
'b',
),
'__header_end__': ' ', # default is '\n'
'__sep__': ' +',
'__line_end__': ' ', # default is '\n'
'__section_end__': ';', # default is ''
'__header_ind__': 2, # default is 0
'__body_ind__': 4, # default is 2 if is header else 0
'__do_upper_keywords__': False, # default is True
}
)
You could add placeholder in query by adding $variable_name syntax.
from sqlconstructor import SqlQuery, SqlContainer
def get_product_query(
product_quality: str,
brand_identifier: int,
) -> SqlContainer:
q = SqlQuery()
q['select'](
'id',
'name',
)
q['from'](
'product'
)
q['where'](
'quality = $quality',
'and brand_id = $brand_id'
)(quality=product_quality, brand_id=brand_identifier)
q['order by']('name DESC')
container: SqlContainer = q()
return container
from sqlconstructor import SqlQuery, SqlContainer
def main():
container: SqlContainer = get_product_query()
# set variables to existing container
container(quality='Best', brand_id=1)
# or
container.vars['quality'] = 'Best'
container.vars['brand_id'] = 1
# if you would like to rewrite all vars
new_vars = {'quality': 'Medium', 'brand_id': 2}
container.vars = new_vars
# if you would like to remove all vars
container.vars.clear()
def get_product_query() -> SqlContainer:
q = SqlQuery()
q['select'](
'id',
'name',
)
q['from'](
'product'
)
q['where'](
'quality = $quality',
'and brand_id = $brand_id'
)
container: SqlContainer = q()
return container
from sqlconstructor import SqlContainer
from functools import cache
def main():
# you could set default values of variables inside of cached result
# and reassign them later.
# Or do not set them in cached result at all and set them later.
container: SqlContainer = get_product_query()
# set/reassign variables to existing container
container(quality='Best', brand_id=1)
@cache
def get_product_query() -> SqlContainer:
...
Use 'dumps' method of SqlContainer. Notice that each value of variables will be conveted to sql by SqlVal class. If you would like to insert value of variable 'as is' in sql string then save variable as SqlContainer.
from sqlconstructor import SqlContainer
from functools import cache
def main():
container: SqlContainer = get_product_query()
container.vars.update(
{
'quality': 'Best', # will be converted by SqlVal to 'Best' (with surrounding single quotes)
'avg': SqlContainer('price'), # will be converted by 'str' method and not by SqlVal (because of SqlContainer)
}
)
# to replace placeholders by variables call 'dumps' method
sql_text: str = container.dumps()
@cache
def get_product_query() -> SqlContainer:
...
If you would like to get sql without replacing placeholders then call '__str__' method of SqlContainer instead of 'dumps':
from sqlconstructor import SqlContainer
from functools import cache
def main():
container: SqlContainer = get_product_query()
container.vars.update({'quality': 'Best', 'brand_id': 1})
# get sql without replacing placeholders by variables
sql_text: str = str(container)
@cache
def get_product_query() -> SqlContainer:
...
You could use & or | operator between filters or betweem filter and str (or any object with str method) (in release >= 1.1.0). Result of & or | operator is SqlContainer and SqlContainer (and SqlWrap) also can operate & and | (in release >= 1.2.0)
from sqlconstructor import SqlQuery, SqlFilter, SqlContainer, SqlWrap
def get_product_query() -> SqlContainer:
q = SqlQuery()
q['select'](
'id',
'name',
)
q['from']('product')
q['where'](
'id <> $identifier' # add placeholder in string
& # AND operator
SqlContainer('price < 10')
&
SqlFilter(
brand_id=SqlPlaceholder('brand_id') # add placeholder as value (you could insert SqlPlaceholder anywhere where value is expected)
) # will be converted by SqlVal to brand_id=$brand_id
&
SqlFilter('quantity > 0') # string will not be converted by SqlVal and will be passed as is
&
SqlWrap(
SqlFilter(quality='Best') # value of the keyword argument will be converted by SqlVal to sql string
| # OR operator
SqlFilter({'rating': 'high'}) # dict value will be converted by SqlVal to sql string
)
)
container: SqlContainer = q()
container(identifier=2, brand_id=1) # set variables after building query for placeholders
return container
Output of str(container)
SELECT
id,
name
FROM
product
WHERE
id <> $identifier
AND
price < 10
AND
brand_id=$brand_id
AND
quantity > 0
AND
(
quality='Best'
OR
rating='high'
)
SqlFilters has dict behavior (in release >= 1.1.4). You could set, get, iterate SqlFilters as dict.
from sqlconstructor import SqlFilters
filters = SqlFilters()
filters['a'] = 1
filters['b'] = 2
one = filters['a']
if 'a' in filters:
del filters['a']
for key, value in filters.items():
...
You could use SqlFilters if all filters require same operator
from sqlconstructor import SqlFilters, AND, OR
# AND mode is default
SqlFilters(
{
'quality': product_quality, # each value of dict will be converted by SqlVal
'brand_id': brand_identifier,
}
)
# explicit AND mode
SqlFilters(
{
'quality': product_quality,
'brand_id': brand_identifier,
},
AND,
)
# OR mode
SqlFilters(
{
'quality': product_quality,
'brand_id': brand_identifier,
},
OR,
)
# Build filters by keyword arguments
# AND as default
SqlFilters(
quality=product_quality,
brand_id=brand_identifier,
)
# OR
SqlFilters(
None,
OR,
quality=product_quality,
brand_id=brand_identifier,
)
You could make query as nested as you would like to.
from sqlconstructor import SqlQuery, SqlContainer
from typing import List
def main():
q = SqlQuery()
q['select'](
'p.id',
'p.name',
'exp.exp_date'
)
q['from'](
'product as p',
)
q['left join lateral'](
get_left_join_lateral(),
)
q['where'](
'p.quality = $quality',
'and p.brand_id = $brand_id'
)(quality='Best', brand_id=1)
def get_left_join_lateral() -> SqlContainer:
j = SqlQuery()
j['select'](
'e.id',
'e.expiration_date as exp_date',
)
j['from']('expiration as e')
j['where'](*get_filters())
j['limit'](100)
"""
You could get SqlContainer with wrapped subquery
in some different ways:
# return r('AS exp ON TRUE')
or
# return r(wrap='AS exp ON TRUE')
"""
# or more explicit
return j().wrap('AS exp ON TRUE')
def get_filters() -> List[str]:
"""Create filters"""
where = []
where.append('p.id = e.id')
where.append('AND e.expiration_date <= now()')
return where
If you pass nested dict in main query dict then it will be subquery. If you add ('__do_wrap__': True) to nested dict then nested subquery will be wrapped by parenthesis. If you add ('__wrapper_text__': any string) to nested dict then nested subquery will be wrapped and wrapper_text will be added after parenthesis (even if you do not add (__do_wrap__: True)).
from sqlconstructor import SqlQuery
q = SqlQuery(
{
'select': (
'p.id',
'p.name',
'exp.exp_date'
),
'from': 'product as p',
'left join lateral': {
'select': (
'e.id',
'e.expiration_date as exp_date',
),
'from': 'expiration as e',
'where': (
'p.id = e.id',
'AND e.expiration_date <= now()',
),
'__wrapper_text__': 'as exp on true',
},
'where': (
"quality = 'Best'",
'and brand_id = 1',
),
}
)
SqlCte is StringConvertible and ContainerConvertible (in release >= 1.3.0).
from sqlconstructor import SqlQuery, SqlContainer, SqlCte
def get_ctes() -> SqlContainer:
"""
Build ctes
"""
ctes = SqlCte()
# regiter cte and fill it later
a: SqlQuery = ctes.reg('warehouse_cte')
a['select'](
'id',
'quantity',
)
a['from']('warehouse')
a['where'](
'id = $id',
'AND quantity > $quantity',
)(id=11, quantity=10)
return ctes()
from sqlconstructor import SqlQuery, SqlContainer, SqlCte
def get_ctes() -> SqlContainer:
"""
Build ctes
"""
ctes = SqlCte()
ctes['warehouse_cte'] = get_warehouse_cte()
# or
# ctes.reg('warehouse_cte', get_warehouse_cte())
# or by dict
# ctes.reg('warehouse_cte', {'select': ('id', 'quantity',) ...})
# you could also get certain cte by name and append new SqlSection to it
a = ctes['warehouse_cte']
a['limit'](1)
return ctes()
def get_warehouse_cte() -> SqlQuery: # or SqlContainer
a = SqlQuery()
a['select'](
'id',
'quantity',
)
a['from']('warehouse')
a['where'](
'id = $id',
'AND quantity > $quantity',
)(id=11, quantity=10)
return a # or a()
It is so easy!
from sqlconstructor import SqlQuery, SqlContainer
def main():
q = SqlQuery()
q += get_ctes()
q['select'](
'id',
'name',
)
...
def get_ctes() -> SqlContainer:
...
or create and add cte to query in one step in release >= 1.3.13
from sqlconstructor import SqlQuery, SqlContainer
def main():
q = SqlQuery()
q.ctes['bestsellers'] = {'select': ('id', 'sold',) ...}
# or set by SqlContainer or SqlQuery
# q.ctes['bestsellers'] = get_warehouse_cte()
...
In release >= 1.2.5
from sqlconstructor import SqlQuery, SqlContainer
def main():
q = SqlQuery(
{
'products': {
'__is_cte__': True,
'select': 'product_id',
'from': 'warehouse',
'where': 'quantity > 0',
},
'select': (
'id',
'name',
),
},
...
)
...
In release >= 1.3.13 we add extra syntax
from sqlconstructor import SqlQuery, SqlContainer
def main():
q = SqlQuery(
{
'__ctes__': {
'products': {
'select': 'product_id',
'from': 'warehouse',
'where': 'quantity > 0',
},
'product_descriptions': {
'select': 'description',
...
},
},
'select': (
'id',
'name',
),
}
...
)
...
In release >= 1.3.13
from sqlconstructor import SqlQuery
q = SqlQuery()
q.add(
{
'__ctes__': {
'products': {
'select': 'product_id',
'from': 'warehouse',
'where': 'quantity > 0',
},
},
},
)
In release >= 1.0.29 you could enumerate columns and values a little bit easier:
import uuid
from sqlconstructor import SqlQuery, SqlCols, SqlVals
q = SqlQuery()
_uuid = uuid.uuid4()
q['insert into'](
'product',
SqlCols(
'brand_id',
'name',
'quality',
'uuid_id',
),
)
q['values'](
SqlVals(
1,
'phone',
'Best',
_uuid,
)
)
sql_text = str(q())
sql_text output will be
INSERT INTO
product
(
"brand_id",
"name",
"quality",
"uuid_id"
)
VALUES
(
1,
'phone',
'Best',
'82611533-25c4-4cbd-8497-3f5024ca29a1'
)
SqlVals converts python objects to sql values (same mechanics used in '$value' replacement by 'dumps' method of SqlContainer).
If you would like to do not add double quotes to columns then you could use SqlEnum class. SqlEnum converts to strings as is (without extra processing), and do not add any extra characters (no single quotes either).
Any of this class (SqlEnum, SqlVals, SqlCols) has 'inline' and 'multiline' method (return SqlContainer which you could wrap by 'wrap' method or do not wrap) in release >= 1.0.29. Example:
import uuid
from sqlconstructor import SqlQuery, SqlEnum, SqlVals
q = SqlQuery()
_uuid = uuid.uuid4()
q['insert into'](
'product',
SqlEnum(
'brand_id',
'name',
'quality',
'uuid_id',
).inline().wrap(),
)
q['values'](
SqlVals(
1,
'phone',
'Best',
_uuid,
).inline().wrap()
)
sql_text = str(q())
sql_text output will be
INSERT INTO
product
(brand_id, name, quality, uuid_id)
VALUES
(1, 'phone', 'Best', '82611533-25c4-4cbd-8497-3f5024ca29a1')
SqlEnum, SqlCols, SqlVals classes have list behavior (in release >= 1.1.4). You could set, get, iterate any of theses classes as list:
from sqlconstructor import SqlVals
sql_vals = SqlVals('a', 'b')
sql_vals[0] = 'f'
f_char = sql_vals[0]
sql_vals.append('c')
sql_vals.extend(['d', 'e'])
sql_vals.pop()
if 'd' in sql_vals:
sql_vals.remove('d')
It is possible to insert single sql value and column by SqlVal and SqlCol class instance respectively (in release >= 1.1.0).
from sqlconstructor import SqlQuery, SqlCol, SqlVal
# create SqlQuery instance
q = SqlQuery()
# register as many SqlSection instances as you'd like
q['select'](
SqlCol('id'),
SqlCol('name'),
)
q['from'](
'product'
)
q['where'](
'quality = ' + SqlVal('Best'),
f"and brand_id = {SqlVal(1)}",
)
SqlContainer inherits all vars of another SqlContainer if it provided as argument in construction (in release >= 1.1.8). You could add inline wrap if you provide 'do_multiline=False' argument in 'wrap' method (in release >= 1.1.8). Multiline type of wrapping is default.
SqlContainer has 'indent' method to make positive or negative indentation (in release >= 1.2.5).
It is possible wrap any str or string convertible object by SqlWrap (in release >= 1.1.1). SqlWrap also could operate & or | as SqlFilter (in release >= 1.1.4). SqlWrap is subclass of SqlContainer (in release >= 1.1.8). SqlWrap has inline and multiline methods (in release >= 1.1.8). It only change type of wrapping and do not make whole text in one line (or in multi lines).
from sqlconstructor import SqlFilter, SqlWrap
result = str(SqlWrap(SqlFilter(a=1) & SqlFilter(b=2)))
result is:
(
a=1
AND
b=2
)
In release >= 1.4.0 we add coalesce and nullif in python implementation
from sqlconstructor import coalesce, nullif, SqlVal
a = coalesce(None, 3, 4, None) # will be 3
b = coalesce(None, None) # will be None
c = SqlVal(coalesce(None, None)) # will be 'null'
d = nullif(3, 4) # will be 3
e = nullif(3, 3) # will be None
f = SqlVal(nullif(3, 3)) # will be 'null'
In release 1.4.1 we add SqlCase class. It is invented to be python implementation of CASE-WHEN-THEN-ELSE-END construction. Each WHEN-THEN is a tuple. ELSE statement is single value. After ELSE statement construction will be closed even if you add tuples after it.
Example of SqlCase use:
from sqlconstructor import SqlCase
sql_case = SqlCase(('b=1', "'one'"), ('b=2', "'two'"), "'nothing'")
result of str(sql_case) will be:
CASE
WHEN
b=1
THEN
'one'
WHEN
b=2
THEN
'two'
ELSE
'nothing'
END
SqlQuery is StringConvertible (in release >= 1.3.0). Now it supports __add__ and __radd__ methods with another StringConvertible subclasses.
__add__ and __radd__ returns SqlContainer and inherits vars of any SqlContainer instance arg.
StringConvertible subclasses:
You could convert each instance of classes below into SqlContainer and set sql variables in one step by __call__ method (in release >= 1.1.5): SqlCol, SqlCols, SqlCte, SqlEnum, SqlFilter, SqlFilters, SqlPlaceholder, SqlSectionHeader, SqlVal, SqlVals, SqlCte (in release >= 1.3.0).
SqlJson (in release >= 1.2.9) has 'loads' and 'dumps' static methods similar to python 'json' library. 'loads' convert from sql json to python object and 'dumps' convert from python object to sql json.
If you would like to find your piece of code in editor by ready sql which is produced by sqlconstructor then you have to mark SqlQuery instances by 'sql_id' parameter in advance (before you have produced ready sql):
from sqlconstructor import SqlQuery, SqlContainer
def main():
q = SqlQuery()
q += get_part_of_query()
q['select'](
'id',
'name'
)
...
def get_part_of_query() -> SqlContainer:
p = SqlQuery(sql_id='25b11c69-ae05-4804-89ea-8ee405f6be8b')
p['select']('quantity')
...
Since you added sql_id, now the SqlQuery instance has such value in sql_id attribute and it will add comment to produced sql as:
-- sql_id='25b11c69-ae05-4804-89ea-8ee405f6be8b'
SELECT
quantity
...
Now when you see sql_id in your logs then it will be easy to find that part of code in your editor!
FAQs
Let you build SQL requests in pythonic way. SQL blocks could be nested and flexible as you want it to be, constructed dynamically and look pretty! You could also build request once and cache it (but still use variables to replace dynamically). Solid solution!
We found that sqlconstructor demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Security News
vlt's new "reproduce" tool verifies npm packages against their source code, outperforming traditional provenance adoption in the JavaScript ecosystem.
Research
Security News
Socket researchers uncovered a malicious PyPI package exploiting Deezer’s API to enable coordinated music piracy through API abuse and C2 server control.
Research
The Socket Research Team discovered a malicious npm package, '@ton-wallet/create', stealing cryptocurrency wallet keys from developers and users in the TON ecosystem.