A simple sql builder based on standard Python type hints
Project description
Sqlify
This project is a fork from pg_simple, that tries to implement a standard SQL with python type hinting interface. This fork also implements extra parameters like having or with queries. Other goals for this project is to support other types of databases like sqlite.
The Sqlify module provides a simple standardized interface while keeping the
benefits and speed of using raw queries over psycopg2
or sqlite3
This module is ment to work as a query builder, and you must provide your own integrations and session pooling if you want.
sqlify
is not intended to provide ORM-like functionality, rather to make it easier to interact with the database from
python code for direct SQL access using convenient wrapper methods.
The sqlify
module provides:
- Python typed interface that can scale from just basic queries to some complex queries, for example using the PostgreSQL With
- Python API to wrap basic SQL functionality: select, update, delete, join et al
- Query results as python dict objects
- Inserts/Updates/Deletes returning data as dict objects or the affected rows count
- Auto commit/rollback when finishing one or multiple queries
- Database migration tools
- Typer cli for migration commands
- Bulk insert (WIP)
- On the fly error prevention when developing with a smart IDE like pycharm (due to the advanced type hinting)
- Debug logging support
Installation
With pip
or easy_install
:
pip install sqlify
or:
easy_install sqlify
or from the source:
python setup.py install
30 Seconds Quick-start Guide
- Step 1: Connect to the database of your choice
- Step 2: Using the Session class pass through the connection
- Step 3: Enjoy your queries
Here's a pseudo-example to illustrate the basic concepts:
import sqlite3
from sqlify import Session
conn = sqlite3.connect('my_test.db')
with Session(conn, autocommit=True) as sqlify:
rest = sqlify.fetchone(
table="test",
fields="column_1",
)
Basic Usage
Connecting to the posgtresql server
The following snippet will connect to the posgtresql server and allocate a cursor:
import psycopg2
from sqlify import Session
conn = psycopg2.connect("host=localhost dbname=test user=postgres password=postgres")
with Session(conn, autocommit=True) as sqlify:
rest = sqlify.fetchone(
table="test",
fields="column_1",
)
By default psycopg2
generates result sets as collections.namedtuple
objects (using psycopg2.extras.NamedTupleCursor
).
But because sqlify
is connection agnostic you can easily modify it to use the DictCursor
that returns a Dict
object
import psycopg2
from psycopg2.extras import DictCursor
conn = psycopg2.connect("host=localhost dbname=test user=postgres password=postgres", cursor_factory=DictCursor)
If you don't like context based interfaces (aka with statement) or it doesn't fit your architecture you can also assign it to a variable and use it as you did expect. But remember that by using it this way you lost the auto-commit/rollback feature and auto-close of the database connection
sqlify = Session(conn, autocommit=True).session
rest = sqlify.fetchone(
table="test",
fields="column_1",
)
sqlify.commit()
sqlify.close()
Fetching a single record
with Session(conn, autocommit=True) as sqlify:
book = sqlify.fetchone(
table='books',
fields="*",
where=(
"published = %(publish_date)s",
dict(
publish_date=datetime.date(2002, 2, 1),
),
),
)
print(f"{book.name} was published on {book.published}")
Fetching multiple records
from sqlify import Order
with Session(conn, autocommit=True) as sqlify:
books = sqlify.fetchone(
table='books',
fields=['name AS n', 'genre AS g'],
where=(
[
"published BETWENN %(since)s and %(to)s",
"gender = %(gender)s",
],
dict(
since=datetime.date(2005, 2, 1),
to=datetime.date(2009, 2, 1),
gender="fiction",
),
),
order=("published", Order.DESC),
limit=5,
offset=2,
)
for book in books:
print(f"{book.name} was published on {book.published}")
Raw SQL execution
In raw queries you can use both list
and dict
annotations
with Session(conn, autocommit=True) as sqlify:
sqlify.execute('SELECT tablename FROM pg_tables WHERE schemaname=%s and tablename=%s', ['public', 'books'])
sqlify.execute('SELECT name FROM books WHERE author=%(author)s', {"author": "Andre"})
Inserting rows
with Session(conn, autocommit=True) as sqlify:
for i in range(1, 10):
sqlify.insert(
table="books",
data=dict(
name=f"Book Name vol. {i}",
price=1.23 * i,
genre="fiction",
published=f"{2000 + i}-{i}-1",
),
)
# DB commit is already called when the session context exits without any exception
# You can disable this with autocommit=False
Updating rows
from sqlify import RawSQL
with Session(conn, autocommit=True) as sqlify:
affected_rows = sqlify.update(
table="books",
where=(
"published = %(published)s",
dict(
published=datetime.date(2001, 1, 1)
),
),
data=dict(
genre="non-fiction",
modified=RawSQL("now()"),
),
)
# Commit is implicit
print(f"Lines updated in this query: {affected_rows}")
Deleting rows
with Session(conn, autocommit=True) as sqlify:
deleted_rows = sqlify.delete(
table="books",
where=(
"published >= %(published)s",
dict(published=datetime.date(2005, 1, 31)),
),
)
# Commit is implicit
print(f"Lines deleted in this query: {deleted_rows}")
Dropping and creating tables
with Session(conn, autocommit=True) as sqlify:
sqlify.drop('books')
sqlify.create('books',
'''
"id" SERIAL NOT NULL,
"type" VARCHAR(20) NOT NULL,
"name" VARCHAR(40) NOT NULL,
"price" MONEY NOT NULL,
"published" DATE NOT NULL,
"modified" TIMESTAMP(6) NOT NULL DEFAULT now()
'''
)
sqlify.execute('''ALTER TABLE "books" ADD CONSTRAINT "books_pkey" PRIMARY KEY ("id")''')
# Commit is implicit
Emptying a table or set of tables
with Session(conn, autocommit=True) as sqlify:
sqlify.truncate('tbl1')
sqlify.truncate('tbl2, tbl3', restart_identity=True, cascade=True)
# Commit is implicit
Inserting/updating/deleting rows with return value
with Session(conn, autocommit=True) as sqlify:
row = sqlify.insert(
table="books",
data=dict(
name=f"Book Name vol. {i}",
price=1.23 * i,
genre="fiction",
published=f"{2000 + i}-{i}-1",
),
returning="id",
)
print(row.id)
with Session(conn, autocommit=True) as sqlify:
rows = sqlify.update(
table="books",
where=(
"published = %(published)s",
dict(
published=datetime.date(2001, 1, 1)
),
),
data=dict(
genre="non-fiction",
modified=RawSQL("now()"),
),
returning="*",
)
for row in rows:
print(row.modified)
with Session(conn, autocommit=True) as sqlify:
rows = sqlify.delete(
table="books",
where=(
"published >= %(published)s",
dict(published=datetime.date(2005, 1, 31)),
),
returning="name",
)
for row in rows:
print(row.name)
Explicit database transaction management
with Session(conn, autocommit=False) as sqlify:
try:
sqlify.execute('Some SQL statement')
sqlify.commit()
except:
sqlify.rollback()
Implicit database transaction management
with Session(conn, autocommit=True) as sqlify:
sqlify.execute('Some SQL that trows an error')
# Rollback will automatically be called and the exception will continue down the execution tree
The above transaction will be rolled back automatically should something goes wrong.
Project details
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.