SQLBatis¶

SQLBatis provides an easier way to interact with the database through the raw sql.

Requirements¶

Installation¶

Install SQLBatis with command pip:

pip install sqlbatis

Tutorial¶

Setup your DB first¶

SQLBatis provides the ability to interact with the database through command-line base on the alembic. There is sqlbatis command is available once the sqlbatis is installed, you can use it directly.

Four common commands will be mentioned here, we will take a quick tour about them, also create a database for your app from scratch step by step, and you can run the command:

sqlbatis --help

to get more details about sqlbatis

Define a Model¶

SQLBatis can mapping the model to the table in the database, but there are some rules that we need to follow, Let’s define a Model first:

from sqlalchemy import Column, Integer, String
from sqlbatis import Model

class User(Model):

    id = Column(Integer, primary_key=True)
    name = Column(String)
    full_name = Column(String)

All the Models should inherit from sqlbatis.model.Model, so we can find all the models that you defined in the app, and create them in your database. We will convert the class name to camel case when we mapping the class to the table, and use it as table name.But if __tablename__ is specified we will use it instead of the camel case:

from sqlalchemy import Column, Integer, String
from sqlbatis import Model

class User(Model):

    # will use 'customized_table_name' instead of 'user'
    __tablename__ = 'customized_table_name'

    id = Column(Integer, primary_key=True)
    name = Column(String)
    full_name = Column(String)

tips: Column, Integer, String etc. are imported from the sqlalchemy

Scan all the models¶

You can use the command:

sqlbatis scan

to check the models that you defined in the app, the useage of this command:

Usage: sqlbatis scan [OPTIONS]

Show all the models that searched in the directory

Options:
-d, --directory TEXT  The main folder that you want to search the models,
                        default is current work directory
-i, --ignore TEXT     Ignore the folder or files through defined regrex
                        expression, List[regrex]
--help                Show this message and exit.

now the default ignore directories are:

['tests', 'build', 'vscode','dist', 'egg', 'migrations', 'sqlbatis']

Initialize the tools¶

Once you confirmed the models that you defined, just use the command:

sqlbatis init

to generate the initial configuration files which will be used in migrate and upgrade stage. the usage of this:

Usage: sqlbatis init [OPTIONS]

Init the db tools

Options:
-d, --directory TEXT  The main folder that you want to search the models,
                        default is current work directory
-i, --ignore TEXT     Ignore the folder or files through defined regrex
                        expression, List[regrex]
-db, --db_url TEXT    The database url config
--help                Show this message and exit.

This command will generate several configuration files in the migrations folder automatically, the migrate scripts will be generated according to the configuration defined here.

Caution: We need to specify the -db when we execute this command, will remove that in the near future.

Generate the scripts¶

Here we will generate the scripts for upgrade, the command:

Usage: sqlbatis migrate [OPTIONS]

Generate the migrate script

Options:
--help  Show this message and exit.

The upgrade script will be generated in the migrations/versions/ folder, and you can find the script name from the console. Before you go to next step, you still have the opportunity to modify the scripts in the versions folder utils you think it’s make sense.

Sync the DB¶

We have not mapping the models to our database table util the upgrade command is executed:

Usage: sqlbatis upgrade [OPTIONS]

Upgrade the db to the version specified, if not sepecified will update to
the latest version

Options:
-v, --version TEXT  The version that you want to upgrade to.
--help              Show this message and exit.

Now, the table will be created in your DB.

Let’s talk with DB in your app¶

Connect to the DB¶

The first thing that we need to do is initialize the sqlbatis instance with DB url, so the tool can know which database should be connected. The code snippet below:

from sqlbatis import SQLBatis
db = SQLBatis('sqlite:///:memory:')

To have a quick start, we use the sqlite as an example.

Actually the SQLBatis will pass all the parameters to the SQLAlchemy create_engine function, you can define the extra create options defined in the SQLAlchemy.

Execute the raw sql with @db.query¶

tips: Explanation first, the db is the instance of SQLBatis, the same definition in the below sections

We have provided the decorator @db.query - sqlbatis.sqlbatis.SQLBatis.query() to execute the raw sql, Hera are CRUD examples:

from sqlalchemy import Column, Integer, String
from sqlbatis import SQLBatis, Model
db = SQLBatis('sqlite:///:memory:')

class User(Model):

    id = Column(Integer, primary_key=True)
    name = Column(String)
    full_name = Column(String)


@db.query('INSERT INTO user (name, full_name) VALUS(:name, :full_name)')
def create(name, full_name):
    pass

@db.query('SELECT * FROM user')
def query_user():
    pass

@db.query('UPDATE user SET name = :name WHERE id = :id')
def update_user(name, id):
    pass

@db.query('DELETE FROM user WHERE id = :id')
def delete_user(id):
    pass

if __name__ == '__main__':
    create('10111000', 'Leo')

As you can see here, although it is named query, you still can execute the insert, update and delete statement. the parameters defined in the query is named style in the Python DBAPI, you can pass the parameters to the function, and the decorator will retrieve those values mapping with the parameters defined in the statement.

Also, you can use a dict instead of the positional arguments.:

if __name__ == '__main__':
    dic = {
        'name': '10111000',
        'full_name': 'Leo'
    }
    create(dic)

The Results of the query is the RowSet object which defined in the SQLBatis, Please see the more details in the API Reference - sqlbatis.row.RowSet.

Insert or update bulk records¶

In some scnarios, we need to do the bulk insert or update, actually, it is more efficient to do bulk operations rather than do it one by one. This is a suitable work for the @db.bulk_query - sqlbatis.sqlbatis.SQLBatis.bulk_query():

BULK_INSERT_SQL = 'INSERT INTO user (name, full_name) VALUS(:name, :full_name)'

# users are the list of the user
users = [{'name': '10111000', 'full_name': 'Leo'} for _ in range(16)]

@db.bulk_query(BULK_INSERT_SQL)
def bulk_insert(users):
    pass

The main difference between the @db.query and @db.bulk_query, is the parameter that we pass to the function is the list of dictionaries.

Paginate query with SQLBatis¶

@db.query_by_page - sqlbatis.sqlbatis.SQLBatis.query_by_page() is a helper function to do the pagination. you can easier to do the paging query with this, the useage below:

@db.query_by_page('SELECT * FROM user', page=1, page_size=10)
def query_user_by_page():
    pass

this decorator will recieve three parameters - sql, page, page_size. It is easy to understand what does it means. will not explain it any more.

In terms of the wrong page or page size, the default process is:

if the page size less than 1, will set the page size to 10

if page less than 1, it means the 0 or negative page number, will set 1

if page greater than max page, will set the page number to the max page number

The result of query_by_page is the sqlbatis.page_query_builder.PageResults. Please get more details in API documentation.

Execute the Query in a transaction¶

sqlbatis.sqlbatis.transactional() is a decorator which help you to do the query in a transaction:

@db.transactional()
def query_in_transaction():
    create(user)
    raise Exception('transaction error')
    create(user)

Due to the exception raised in the function, the transaction will not be committed to the database, the first create function will be rolled back.

In terms of the case below:

@db.transactional()
def transaction_outer_for_inner_exception():
    create(user)
    create(user)
    transaction_inner_for_inner_exception()


@db.transactional()
def transaction_inner_for_inner_exception():
    create(user)
    raise Exception('transaction error')

the error raised in the inner function, will cause the db operations roll back in the outer function. so nothing will be committed.

Builtin Database Access Operations¶

The most common operations when we interact with a database are CRUD, SQLBatis also provides these basic functions for your usage:

CRUD operations¶

create: will create a row in the database, sqlbatis.sqlbatis_dao.SQLBatisDao.create()
retrieve_by_id: will get a row according to the id which is also as required parameter need to pass to the function, sqlbatis.sqlbatis_dao.SQLBatisDao.retrieve_by_id():
retrieve_all: get all the rows in the table. sqlbatis.sqlbatis_dao.SQLBatisDao.retrieve_all()
update_by_id: Update the row in the database by the id and attrs. sqlbatis.sqlbatis_dao.SQLBatisDao.update_by_id()
delete_by_id: Delete the row by the primary key. sqlbatis.sqlbatis_dao.SQLBatisDao.delete_by_id()
filter_by: Filter the rows according to the dictionary we pass to the func. sqlbatis.sqlbatis_dao.filter_by()

Bulk insert¶

bulk_insert: provide a way to insert the multiple rows and update multiple rows in the database. sqlbatis.sqlbatis_dao.SQLBatisDao.bulk_insert()

Reference¶

API¶

sqlbatis.cli¶

Provide the cli to interact with DB through command-line

Usage: sqlbatis [OPTIONS] COMMAND [ARGS]…

Options:

--help

Show this message and exit.

Commands:

scan Show all the models that searched in the directory

init Init the db tools

migrate Generate the migrate script

upgrade Upgrade the db to the version specified, if not sepecified will update to the latest version

downgrade: Downgrade the db to the version sepecified, need to give the version parameter

sqlbatis.connection¶

class sqlbatis.connection.Connection(conn)[source]¶

Bases: object

The wrapper of the sqlalchemy raw connection

begin()[source]¶

Start a transaction

Returns:	a transaction
Return type:	TBI

begin_nested()[source]¶

bulk_query(sql, *params)[source]¶

Bulk update or insert

Parameters:	sql (str) – the raw sql that will be executed
Returns:	the row or rowset of the sql result
Return type:	Row or RowSet

close()[source]¶: Close the connection

closed¶

Check the connection status

Returns:	the boolean which indicate the connection status
Return type:	boolean

execute(sql, fetch_all=False, inserted_primary_key=False, **params)[source]¶

The raw execute function of the sqlalchemy, and main difference between this func with the query is it can accept the sqlalchemy sql expression as the first parameter

Parameters:	sql (sqlalchemy sql expression) – the sqlalchemy sql expression need to be executed fetch_all (bool, optional) – determine if consume all the iterator immediately instead of lazy loading, defaults to False inserted_primary_key (bool, optional) – if return the primary key when do the create func, defaults to False
Returns:	the result of the query
Return type:	Row or RowSet or int(if inserted_primary_key)

in_transaction()[source]¶

Check if the connection is in transaction

Returns:	the status of the connection transaction
Return type:	boolean

query(sql, fetch_all, **params)[source]¶

The basic query based on the sqlalchemy, it will accept the raw sql, and execute will raw sqlalchemy connection

Parameters:	sql (str) – the raw sql that will be executed fetch_all (bool) – determine if consume all the iterator immediately instead of lazy loading
Returns:	the row or rowset of the sql result
Return type:	Row or RowSet

sqlbatis.container¶

class sqlbatis.container.SQLBatisContainer[source]¶

Bases: object

The container for the SQLBatis injection, it will hold the instance that class which inherit the SQLBatisMetaClass

Raises:	ContainerException – will raise exception if no instance is registered

static get(key)[source]¶

Try to get the instance from the container

Parameters:	key (str) – the key of the instance
Raises:	ContainerException – if no instance is binded with the key, will raise the exception
Returns:	the instance
Return type:	cls instance

static has_key(key)[source]¶

Check if the instance exist in the container

Parameters:	key (str) – the instance key
Returns:	True if the instance exists else False
Return type:	bool

static register(key, instance)[source]¶

Register the instance to the container

Parameters:	key (str) – the key of the instance instance (cls instance) – the instance of the cls

class sqlbatis.container.SQLBatisMetaClass[source]¶

Bases: type

The mataclass which will register the current instance to the container

be careful: it will just host one instance in the container, the original instance will be replaced if a class instance is generated

sqlbatis.container.entity(cls)[source]¶

Hand over the instance management to the container

Returns:	cls object
Return type:	SQLBatisMetaClass class

sqlbatis.errors¶

exception sqlbatis.errors.ConnectionException[source]¶: Bases: sqlbatis.errors.SQLBatisException

exception sqlbatis.errors.ContainerException[source]¶: Bases: sqlbatis.errors.SQLBatisException

exception sqlbatis.errors.NotSupportedException[source]¶: Bases: sqlbatis.errors.SQLBatisException

exception sqlbatis.errors.PrimaryKeyMissingException[source]¶: Bases: sqlbatis.errors.SQLBatisException

exception sqlbatis.errors.PropagationException[source]¶: Bases: sqlbatis.errors.SQLBatisException

exception sqlbatis.errors.QueryException[source]¶: Bases: sqlbatis.errors.SQLBatisException

exception sqlbatis.errors.SQLBatisException[source]¶: Bases: Exception

exception sqlbatis.errors.SQLInjectionException[source]¶: Bases: sqlbatis.errors.SQLBatisException

exception sqlbatis.errors.TableMissingException[source]¶: Bases: sqlbatis.errors.SQLBatisException

exception sqlbatis.errors.TransactionException[source]¶: Bases: sqlbatis.errors.SQLBatisException

sqlbatis.model¶

class sqlbatis.model.Model[source]¶

Bases: object

Basic Model in the SQLBatis, all the defined model should inherit this class, it will automaticlly interact with SQLAlchemy Core functions:

sqlbatis.page_query_builder¶

class sqlbatis.page_query_builder.PageQueryBuilder(sql, params, page, page_size, fetch_all)[source]¶

Bases: object

Construct a pagination query according to the params that user passed in

get_current_page_results()[source]¶

Construct the sql to get the current page results, and execute

Returns:	The page results
Return type:	Rowsets

get_total_number()[source]¶

Construct the sql to query the total number of rows, and execute

Returns:	the total number of rows
Return type:	int

query()[source]¶

Execute the pagenation query and return the results

Returns:	PageResults object
Return type:	PageResults

class sqlbatis.page_query_builder.PageResults(page, page_size, total, results)[source]¶

Bases: object

The class that contain results of the each page

page: current page number

page_size: number of the object per page

total: total number of query objects

results: the result by page, it is a RowSet object

has_next¶

Check if has next page or not

Returns:	if has next page or not
Return type:	bool

has_prev¶

Check if has previous page or not

Returns:	if has previous page or not
Return type:	bool

sqlbatis.row¶

class sqlbatis.row.Row(columns, values)[source]¶

Bases: object

Row objecct which constructed by the Result retrieved by sqlalchemy

columns()[source]¶

Columns of result

Returns:	all columns
Return type:	list

get(column, default=None)[source]¶

Get the value by the column

Parameters:	column (str) – the column default (any, optional) – default value if the column not exist, defaults to None
Returns:	the value conresponding to column
Return type:	any

to_dict(ordered=False)[source]¶

Return the row as a dictionary, if ordered is True, return an ordereddict

Parameters:	ordered (bool, optional) – if need keep order or not, defaults to False
Returns:	the row dictionary
Return type:	dict

values()[source]¶

Values of result

Returns:	values conrespond to the columns
Return type:	list

class sqlbatis.row.RowSet(rows, result_proxy)[source]¶

Bases: object

Rows collection

all(to_dict=False, to_ordered_dict=False)[source]¶: Fetch all the rows that contained in the rowset, and consume the iterator.

columns()[source]¶

first(default=None, to_dict=False, to_ordered_dict=False)[source]¶: Return the first row in the rowset, and will return the default value if the rowset is empty

is_empty¶

if the empty or not

Returns:	check the empty or not
Return type:	boolean

next()[source]¶

one(default=None, to_dict=False, to_ordered_dict=False)[source]¶: Returns a single row for the RowSet or default value, also we will check there is only one row contained in the RowSet

scalar(default=None)[source]¶: Returns the first column of the first row, or default.

to_dict(ordered=False)[source]¶

class sqlbatis.row.SQLAlchemyResultProxy(result_proxy)[source]¶

Bases: object

inserted_primary_key¶

rowcount¶

sqlbatis.scanner¶

class sqlbatis.scanner.ModelScanner(directory='.', exclude=[])[source]¶

Bases: sqlbatis.scanner.Scanner

scan_models()[source]¶

Scan all the sqlbatis models in the folder that user specified

Returns:	SQLBatis models list
Return type:	list

class sqlbatis.scanner.Scanner(directory, exclude, file_type)[source]¶

Bases: object

The basic scanner to search the files in specific folder.

sqlbatis.sqlbatis¶

class sqlbatis.sqlbatis.SQLBatis(database_url, **kwargs)[source]¶

Bases: object

The basic object to do the query with raw sql

bulk_query(sql)[source]¶

Bulk update or insert with this decorator, it has the similar usage like the query it also have the requirements for the inner function, which means the paramters should be the list of the object that we want to do update or insert

Parameters:	sql (str) – the raw sql that you want to execute

close()[source]¶: Close the sqlbatis, which also mean close the engine of the sqlalchemy

get_connection()[source]¶

The function to get the connection, all the connections are in the localstack object

Raises:	ConnectionException – if the engine is closed, the connection will be created
Returns:	return a connection for query
Return type:	Connection

query(sql, fetch_all=False)[source]¶

The decorator that using for the raw sql query, the simple example for usage is like:

@db.query("SELECT * FROM user")
def get_users():
    pass

then if we try to call the function get_users, the sqlbatis will execute the query automatically.

Parameters:	sql (str) – the sql that you want to execute fetch_all (bool, optional) – will retrieve all the results instead of lazy loading the data from db, defaults to False

query_by_page(sql, page=1, page_size=10, fetch_all=True)[source]¶

Get the rows by page number and page size

Parameters:	sql (str) – The raw SQL that you want to execute page (int, optional) – page number, defaults to 1 page_size (int, optional) – number of rows per page, defaults to 10 fetch_all (bool, optional) – ignore lazy loading or not, defaults to True

transactional()[source]¶

The decorator that for do the transaction, the useage of this is:

@db.transactional()
def transaction_needed_func():
    do(1)
    do(2)

any error occurred, the changes will be rolled back.

also include the nested transaction, consider the scenario like this:

@db.transactional():
def transaction_func_1():
    do(1)
    transaction_func_2()

@db.transactional()
def transaction_func_2():
    do(2) 

if the transaction_func_2 is failed, the result of the do(1) also will rolled back

sqlbatis.sqlbatis_dao¶

class sqlbatis.sqlbatis_dao.SQLBatisDao[source]¶

Bases: object

Basic Dao operations provided by the SQLBatis

bulk_insert(attrs)[source]¶

Bulk insert and update

Parameters:	attrs (list[dict]) – the list of the attributes dict
Returns:	TBI
Return type:	TBI

create(attrs)[source]¶

Insert a row of the object to the DB

Parameters:	attrs (dict) – the value of the columns
Returns:	the primary key of the inserted record
Return type:	int

delete_by_id(_id)[source]¶

Delete the row by the primary key

Parameters:	_id (int) – row primary key
Returns:	TBI
Return type:	TBI

filter_by(where_condition)[source]¶

Get the rows which match the given conditions, if where_condition is empty, will retrieve all rows in the table

Parameters:	where_condition (dict) – conditions that we need to filter from the table
Returns:	rows which filtered by the conditions
Return type:	RowSet

retrieve_all()[source]¶

Get all the rows from the database

Returns:	all the rows in the database
Return type:	RowSet

retrieve_by_id(_id)[source]¶

Get the row by the id

Parameters:	_id (int) – primary key of the row that you want to retrieve
Returns:	the row which id is _id
Return type:	Row

update_by_id(_id, attrs)[source]¶

Update the row in the database by the id and attrs

Parameters:	attrs (dict) – the attributes of row and contained the primary key of the row
Returns:	TBI
Return type:	TBI

sqlbatis.utils¶

sqlbatis.utils.camel_to_snake_case(name, exclude_key_words=[''])[source]¶

tanslate the name to the snake case according to the named conventions 1. split the class name with the capital letter 2. remove all the spaces and join the elements in the array with ‘_’ sign, and lower the string you get

Parameters:	name (str) – the camel case name
Returns:	snake case name
Return type:	str

SQLBatis

Table of Contents

SQLBatis¶

Requirements¶

Installation¶

Tutorial¶

Setup your DB first¶

Define a Model¶

Scan all the models¶

Initialize the tools¶

Generate the scripts¶

Sync the DB¶

Let’s talk with DB in your app¶

Connect to the DB¶

Execute the raw sql with @db.query¶

Insert or update bulk records¶

Paginate query with SQLBatis¶

Execute the Query in a transaction¶

Builtin Database Access Operations¶

CRUD operations¶

Bulk insert¶

Reference¶

API¶

sqlbatis.cli¶

sqlbatis.connection¶

sqlbatis.container¶

sqlbatis.errors¶

sqlbatis.model¶

sqlbatis.page_query_builder¶

sqlbatis.row¶

sqlbatis.scanner¶

sqlbatis.sqlbatis¶

sqlbatis.sqlbatis_dao¶

sqlbatis.utils¶

Dependency injection¶

@entity¶

autowired¶

SQLBatisMetaClass¶

SQLBatisContainer¶

ChangeLog¶

SQLBatis¶

Requirements¶

Installation¶

Tutorial¶

Setup your DB first¶

Define a Model¶

Scan all the models¶

Initialize the tools¶

Generate the scripts¶

Sync the DB¶

Let’s talk with DB in your app¶

Connect to the DB¶

Execute the raw sql with @db.query¶

Insert or update bulk records¶

Paginate query with SQLBatis¶

Execute the Query in a transaction¶

Builtin Database Access Operations¶

CRUD operations¶

Bulk insert¶

Reference¶

API¶

sqlbatis.cli¶

sqlbatis.connection¶

sqlbatis.container¶

sqlbatis.errors¶

sqlbatis.model¶

sqlbatis.page_query_builder¶

sqlbatis.row¶

sqlbatis.scanner¶

sqlbatis.sqlbatis¶

sqlbatis.sqlbatis_dao¶

sqlbatis.utils¶

Dependency injection¶

@entity¶

__autowired__¶

SQLBatisMetaClass¶

SQLBatisContainer¶

ChangeLog¶

autowired¶