Skip to content

API Reference#

SQLAlchemy wrapper class#

class sqla_wrapper.SQLAlchemy(url=None, *, dialect=’sqlite’, name=None, user=None, password=None, host=None, port=None, engine_options=None, session_options=None)

Create a SQLAlchemy connection

This class creates an engine, a base class for your models, and a scoped session.

The string form of the URL is dialect[+driver]://user:password@host/dbname[?key=value..], where dialect is a database name such as mysql, postgresql, etc., and driver the name of a DBAPI, such as psycopg2, pyodbc, etc.

Instead of the connection URL you can also specify dialect (plus optional driver), user, password, host, port, and database name as separate arguments.

Please review the Database URLs section of the SQLAlchemy documentation, for general guidelines in composing URL strings. In particular, special characters, such as those often part of passwords, must be URL-encoded to be properly parsed.

Example:

db = SQLAlchemy(database_uri)
# or SQLAlchemy(dialect=, name= [, user=] [, password=] [, host=] [, port=])

class Base(db.Model):
    pass

class User(Base):
    tablename = "users"
    id = Column(Integer, primary_key=True)
    login = Column(String(80), unique=True)
    deleted = Column(DateTime)
create_all(self, **kwargs)

Creates all the tables of the models registered so far.

Only tables that do not already exist are created. Existing tables are not modified.

drop_all(self, **kwargs)

Drop all the database tables.

Note that this is a destructive operation; data stored in the database will be deleted when this method is called.

test_transaction(self, savepoint=False)

Session class#

class sqla_wrapper.Session(bind=None, autoflush=True, future=False, expire_on_commit=True, autocommit=False, twophase=False, binds=None, enable_baked_queries=True, info=None, query_cls=None)

SQLAlchemy default Session class has the method .get(Model, pk) to query and return a record by its primary key.

This class extends the sqlalchemy.orm.Session class with some useful active-record-like methods and a pagination helper.

all(self, Model, **attrs)

Returns all the object found with these attributes.

The filtering is done with a simple .filter_by() so is limited to “equality” comparisons against the columns of the model. Also, there is no way to sort the results. If you need sorting or more complex filtering, you are better served using a db.select().

Examples:

users = db.s.all(User)
users = db.s.all(User, deleted=False)
users = db.s.all(User, account_id=123, deleted=False)
create(self, Model, **attrs)

Creates a new object and adds it to the session.

This is a shortcut for:

obj = Model(**attrs)
db.s.add(obj)
db.s.flush()

Note that this does a db.s.flush(), so you must later call db.s.commit() to persist the new object.

Example:

new_user = db.s.create(User, email='foo@example.com')
db.s.commit()
first(self, Model, **attrs)

Returns the first object found with these attributes or None if there isn’t one.

The filtering is done with a simple .filter_by() so is limited to “equality” comparisons against the columns of the model. Also, there is no way to sort the results. If you need sorting or more complex filtering, you are better served using a db.select().

Examples:

user = db.s.first(User)
user = db.s.first(User, deleted=False)
first_or_create(self, Model, **attrs)

Tries to find an object and if none exists, it tries to create a new one first. Use this method when you expect the object to already exists but want to create it in case it doesn’t.

This does a db.s.flush(), so you must later call db.s.commit() to persist the new object (in case one has been created).

Examples:

user1 = db.s.first_or_create(User, email='foo@example.com')
user2 = db.s.first_or_create(User, email='foo@example.com')
user1 is user2
create_or_first(self, Model, **attrs)

Tries to create a new object, and if it fails because already exists, return the first it founds. For this to work one or more of the attributes must be unique so it does fail, otherwise you will be creating a new different object.

Use this method when you expect that the object does not exists but want to avoid an exception in case it does.

This does a db.s.flush(), so you must later call db.s.commit() to persist the new object (in case one has been created).

Examples:

user1 = db.s.create_or_first(User, email='foo@example.com')
user2 = db.s.create_or_first(User, email='foo@example.com')
user1 is user2
paginate(self, query, *, total, page=1, per_page=20, padding=0)

Returns a Paginator of the query results. Note that you must calculate the total number of unpaginated results first.

Arguments:

  • query: A select() statement for all items.
  • total: Total number of items. You must pre-calculate this value.
  • page: Number of the current page (first page is 1) It can be a number, a string with a number, or the strings “first” or “last”.
  • per_page: Max number of items to display on each page.
  • padding: Number of elements of the previous and next page to show. For example, if per_page is 10 and padding is 2, every page will show 14 items, the first two from the previous page and the last two for the next one. This extra items will be repeated again on their own pages.

Example:

query = select(User) \
    .where(User.deleted.is_(None)) \
    .order_by(User.created_at)

total = db.s.scalar(
    select(func.count(User.id))
    .where(User.deleted.is_(None))
)

pag = db.s.paginate(query, total=total, page=1, per_page=20)

TestTransaction class#

class sqla_wrapper.TestTransaction(db, savepoint=False)

Helper for building sessions that rollback everyting at the end.

See “Joining a Session into an External Transaction” in the SQLAlchemy documentation.

close(self)

Alembic wrapper class#

class sqla_wrapper.Alembic(db, path=’db/migrations’, **options)

Provide an Alembic environment and migration API.

For a more in-depth understanding of these methods and the extra options, you can read the documentation for the Alembic config.

Arguments:

  • db: A sqla_wrapper.SQLAlchemy instance.
  • path: Path to the migrations folder.
  • **options: Other alembic options
revision(self, message, *, empty=False, parent=’head’)

Create a new revision. Auto-generate operations by comparing models and database.

Arguments:

  • message: Revision message.
  • empty: Generate just an empty migration file, not the operations.
  • parent: Parent revision of this new revision.
upgrade(self, target=’head’, *, sql=False, **kwargs)

Run migrations to upgrade database.

Arguments:

  • target: Revision target or “from:to” range if sql=True. “head” by default.
  • sql: Don’t emit SQL to database, dump to standard output instead.
  • **kwargs: Optional arguments. If these are passed, they are sent directly to the upgrade() functions within each revision file. To use, modify the script.py.makotemplate file so that the upgrade() functions can accept arguments.
downgrade(self, target=’-1’, *, sql=False, **kwargs)

Run migrations to downgrade database.

Arguments:

  • target: Revision target as an integer relative to the current state (e.g.: “-1”), or as a “from:to” range if sql=True. “-1” by default.
  • sql: Don’t emit SQL to database, dump to standard output instead.
  • **kwargs: Optional arguments. If these are passed, they are sent directly to the downgrade() functions within each revision file. To use, modify the script.py.mako template file so that the downgrade() functions can accept arguments.
get_history(self, *, start=None, end=None)

Get the list of revisions in chronological order. You can optionally specify the range of revisions to return.

Arguments:

  • start: From this revision (including it.)
  • end: To this revision (including it.)
history(self, *, verbose=False, start=’base’, end=’heads’)

Print the list of revisions in chronological order. You can optionally specify the range of revisions to return.

Arguments:

  • verbose: If True, shows also the path and the docstring of each revision file.
  • start: Optional starting revision (including it.)
  • end: Optional end revision (including it.)
stamp(self, target=’head’, *, sql=False, purge=False)

Set the given revision in the revision table. Don’t run migrations.

Arguments:

  • target: The target revision; “head” by default.
  • sql: Don’t emit SQL to the database, dump to the standard output instead.
  • purge: Delete all entries in the version table before stamping.
get_current(self)

Get the last revision applied.

current(self, verbose=False)

Print the latest revision(s) applied.

Arguments:

  • verbose: If True, shows also the path and the docstring of the revision file.
get_head(self)

Get the latest revision.

head(self, verbose=False)

Print the latest revision.

Arguments:

  • verbose: If True, shows also the path and the docstring of the revision file.
init(self, path)

Creates a new migration folder with a script.py.mako template file. It doesn’t fail if the folder or file already exists.

Arguments:

  • path: Target folder.
create_all(self)

Create all the tables from the current models and stamp the latest revision without running any migration.

rev_id(self)

Generate a unique id for a revision.

By default this uses alembic.util.rev_id. Override this method to change it.

get_pyceo_cli(self)
get_click_cli(self, name=’db’)
get_flask_cli(self, name=’db’)

Paginator class#

class sqla_wrapper.Paginator(query, *, page=1, per_page=20, total=None, padding=0)

Helper class for paginate data. You can construct it from any iterable.

Arguments:

  • query: Items to paginate.
  • page: Number of the current page (first page is 1) It can be a number, a string with a number, or the strings “first” or “last”.
  • per_page: Max number of items to display on each page.
  • total: Total number of items. If not provided, the length of the iterable will be used.
  • padding: Number of elements of the previous and next page to show. For example, if per_page is 10 and padding is 2, every page will show 14 items, the first two from the previous page and the last two for the next one. This extra items will be repeated again on their own pages.
num_pages

The total number of pages.

total_pages

Alias to num_pages

showing

The number of items in the current page Could be less than per_page if we are in the last page, or more if padding > 0.

is_paginated

True if a more than one page exists.

has_prev

True if a previous page exists.

has_next

True if a next page exists.

next_num

Number of the next page.

prev_num

Number of the previous page.

prev

Returns a Paginator object for the previous page.

next

Returns a Paginator object for the next page.

start_index

0-based index of the first element in the current page.

end_index

0-based index of the last element in the current page.

items

Return the items to for the current page.

pages

Proxy to get_pages()

get_range(self, sep=’ - ‘)

Return a string with the 1-based index range of items in the page (ignoring the padding). Useful for displaying “Showing x - y items of z”.

Examples:

p = Paginator(range(100), per_page=10, page=1)
p.get_range()
'1 - 10'

p = Paginator(range(100), per_page=10, page=5)
p.get_range()
'41 - 50'
get_pages(self, showmax=12)

Return a list of the page numbers in the pagination. The showmax parameter control how many numbers are shown at most.

Depending of the page number and the showmax value, there are several possible scenarios, but the these rules are followed:

  1. The first, last and current pages are always returned.
  2. After those three, the remaining slots are filled around the current page, after the first page, and before the last page, in that order, in turns.
  3. Skipped page numbers are represented as None. We never skip just one page, so the final number of pages shown could be less than the value of showmax.

Examples:

[ (1), 2, 3, 4, 5, 6, None, 10, 11, 12, 13 ]
[ 1, 2, None, 5, 6, 7, (8), 9, 10, None, 13, 14, 15 ]
[ 1, 2, (3), 4, 5 ]

This is one way how you could render such a pagination in the template:

  <p>Showing {{ pg.showing }} or {{ pg.total }}</p>
  <ol class="pg">
  {%- if pg.has_prev %}
    <li><a href="{{ url_for(endpoint, page=pg.prev_num) }}"
     rel="me prev">«</a></li>
  {% else %}
    <li class="disabled"><span>«</span></li>
  {%- endif %}

  {%- for page in pg.pages %}
    {% if page %}
      {% if page != pg.page %}
        <li><a href="{{ url_for(endpoint, page=page) }}"
         rel="me">{{ page }}</a></li>
      {% else %}
        <li class="current"><span>{{ page }}</span></li>
      {% endif %}
    {% else %}
      <li><span class=ellipsis>…</span></li>
    {% endif %}
  {%- endfor %}

  {%- if pg.has_next %}
    <li><a href="{{ url_for(endpoint, page=pg.next_num) }}"
     rel="me next">»</a></li>
  {% else %}
    <li class="disabled"><span>»</span></li>
  {%- endif %}
  </ol>
Back to top