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)

Bases: object

The wrapper of the sqlalchemy raw connection

begin()

Start a transaction

Returns:a transaction Return type:TBI

begin_nested()

bulk_query(sql, *params)

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()

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)

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()

Check if the connection is in transaction

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

query(sql, fetch_all, **params)

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

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)

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)

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)

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

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)

Hand over the instance management to the container

Returns:cls object Return type:SQLBatisMetaClass class

sqlbatis.errors

exception sqlbatis.errors.ConnectionException

Bases: sqlbatis.errors.SQLBatisException

exception sqlbatis.errors.ContainerException

Bases: sqlbatis.errors.SQLBatisException

exception sqlbatis.errors.NotSupportedException

Bases: sqlbatis.errors.SQLBatisException

exception sqlbatis.errors.PrimaryKeyMissingException

Bases: sqlbatis.errors.SQLBatisException

exception sqlbatis.errors.PropagationException

Bases: sqlbatis.errors.SQLBatisException

exception sqlbatis.errors.QueryException

Bases: sqlbatis.errors.SQLBatisException

exception sqlbatis.errors.SQLBatisException

Bases: Exception

exception sqlbatis.errors.SQLInjectionException

Bases: sqlbatis.errors.SQLBatisException

exception sqlbatis.errors.TableMissingException

Bases: sqlbatis.errors.SQLBatisException

exception sqlbatis.errors.TransactionException

Bases: sqlbatis.errors.SQLBatisException

sqlbatis.model

class sqlbatis.model.Model

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)

Bases: object

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

get_current_page_results()

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

Returns:The page results Return type:Rowsets

get_total_number()

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

Returns:the total number of rows Return type:int

query()

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)

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)

Bases: object

Row objecct which constructed by the Result retrieved by sqlalchemy

columns()

Columns of result

Returns:all columns Return type:list

get(column, default=None)

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)

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()

Values of result

Returns:values conrespond to the columns Return type:list

class sqlbatis.row.RowSet(rows, result_proxy)

Bases: object

Rows collection

all(to_dict=False, to_ordered_dict=False)

Fetch all the rows that contained in the rowset, and consume the iterator.

columns()

first(default=None, to_dict=False, to_ordered_dict=False)

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()

one(default=None, to_dict=False, to_ordered_dict=False)

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)

Returns the first column of the first row, or default.

to_dict(ordered=False)

class sqlbatis.row.SQLAlchemyResultProxy(result_proxy)

Bases: object

inserted_primary_key

rowcount

sqlbatis.scanner

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

Bases: sqlbatis.scanner.Scanner

scan_models()

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)

Bases: object

The basic scanner to search the files in specific folder.

sqlbatis.sqlbatis

class sqlbatis.sqlbatis.SQLBatis(database_url, **kwargs)

Bases: object

The basic object to do the query with raw sql

bulk_query(sql)

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()

Close the sqlbatis, which also mean close the engine of the sqlalchemy

get_connection()

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)

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)

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()

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

Bases: object

Basic Dao operations provided by the SQLBatis

bulk_insert(attrs)

Bulk insert and update

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

create(attrs)

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)

Delete the row by the primary key

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

filter_by(where_condition)

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()

Get all the rows from the database

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

retrieve_by_id(_id)

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)

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=[''])

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