How can I add the filter as in SQL to select values that are NOT NULL from a certain column ?

SELECT * 
FROM table 
WHERE YourColumn IS NOT NULL;

How can I do the same with SQLAlchemy filters?

select = select(table).select_from(table).where(all_filters) 

column_obj != None will produce a IS NOT NULL constraint:

In a column context, produces the clause a != b. If the target is None, produces a IS NOT NULL.

or use isnot() (new in 0.7.9):

Implement the IS NOT operator.

Normally, IS NOT is generated automatically when comparing to a value of None, which resolves to NULL. However, explicit usage of IS NOT may be desirable if comparing to boolean values on certain platforms.

Demo:

>>> from sqlalchemy.sql import column
>>> column('YourColumn') != None
<sqlalchemy.sql.elements.BinaryExpression object at 0x10c8d8b90>
>>> str(column('YourColumn') != None)
'"YourColumn" IS NOT NULL'
>>> column('YourColumn').isnot(None)
<sqlalchemy.sql.elements.BinaryExpression object at 0x104603850>
>>> str(column('YourColumn').isnot(None))
'"YourColumn" IS NOT NULL'

Starting in version 0.7.9 you can use the filter operator .isnot instead of comparing constraints, like this:

query.filter(User.name.isnot(None))

This method is only necessary if pep8 is a concern.

source: sqlalchemy documentation

In case anyone else is wondering, you can use is_ to generate foo IS NULL:

>>> from sqlalchemy.sql import column
>>> print column('foo').is_(None)
foo IS NULL
>>> print column('foo').isnot(None)
foo IS NOT NULL

Using SQLAlchemy, an Engine object is created like this:

from sqlalchemy import create_engine
engine = create_engine("postgresql://localhost/mydb")

Accessing engine fails if the database specified in the argument to create_engine (in this case, mydb) does not exist. Is it possible to tell SQLAlchemy to create a new database if the specified database doesn’t exist?

On postgres, three databases are normally present by default. If you are able to connect as a superuser (eg, the postgres role), then you can connect to the postgres or template1 databases. The default pg_hba.conf permits only the unix user named postgres to use the postgres role, so the simplest thing is to just become that user. At any rate, create an engine as usual with a user that has the permissions to create a database:

>>> engine = sqlalchemy.create_engine("postgres://postgres@/postgres")

You cannot use engine.execute() however, because postgres does not allow you to create databases inside transactions, and sqlalchemy always tries to run queries in a transaction. To get around this, get the underlying connection from the engine:

>>> conn = engine.connect()

But the connection will still be inside a transaction, so you have to end the open transaction with a commit:

>>> conn.execute("commit")

And you can then proceed to create the database using the proper PostgreSQL command for it.

>>> conn.execute("create database test")
>>> conn.close()

SQLAlchemy-Utils provides custom data types and various utility functions for SQLAlchemy. You can install the most recent official version using pip:

pip install sqlalchemy-utils

The database helpers include a create_database function:

from sqlalchemy import create_engine
from sqlalchemy_utils import database_exists, create_database

engine = create_engine("postgres://localhost/mydb")
if not database_exists(engine.url):
    create_database(engine.url)

print(database_exists(engine.url))

It’s possible to avoid manual transaction management while creating database by providing isolation_level='AUTOCOMMIT' to create_engine function:

import sqlalchemy

with sqlalchemy.create_engine(
    'postgresql:///postgres',
    isolation_level='AUTOCOMMIT'
).connect() as connection:
    connection.execute('CREATE DATABASE my_database')

Also if you are not sure that database doesn’t exist there is a way to ignore database creation error due to existence by suppressing sqlalchemy.exc.ProgrammingError exception:

import contextlib
import sqlalchemy.exc

with contextlib.suppress(sqlalchemy.exc.ProgrammingError):
    # creating database as above

Please note that I couldn’t get the above suggestions with database_exists because whenever I check if the database exists using if not database_exists(engine.url): I get this error:

InterfaceError(‘(pyodbc.InterfaceError) (\’28000\’, u\'[28000] [Microsoft][SQL Server Native Client 11.0][SQL Server]Login failed for user \\’myUser\\’. (18456) (SQLDriverConnect); [28000] [Microsoft][SQL Server Native Client 11.0][SQL Server]Cannot open database “MY_DATABASE” requested by the login. The login failed. (4060); [28000] [Microsoft][SQL Server Native Client 11.0][SQL Server]Login failed for user \\’myUser\\’. (18456); [28000] [Microsoft][SQL Server Native Client 11.0][SQL Server]Cannot open database “MY_DATABASE” requested by the login. The login failed. (4060)\’)’,)

Also contextlib/suppress was not working and I’m not using postgres so I ended up doing this to ignore the exception if the database happens to already exist with SQL Server:

import logging
import sqlalchemy

logging.basicConfig(filename='app.log', format='%(asctime)s-%(levelname)s-%(message)s', level=logging.DEBUG)
engine = create_engine('mssql+pyodbc://myUser:mypwd@localhost:1234/MY_DATABASE?driver=SQL+Server+Native+Client+11.0?trusted_connection=yes', isolation_level = "AUTOCOMMIT")

try: 
    engine.execute('CREATE DATABASE ' + a_database_name)
except Exception as db_exc:
    logging.exception("Exception creating database: " + str(db_exc))  

Just a quick question: SQLAlchemy talks about calling sessionmaker() once but calling the resulting Session() class each time you need to talk to your DB. For me that means the second I would do my first session.add(x) or something similar, I would first do

from project import Session
session = Session()

What I did until now was to make the call session = Session() in my model once and then always import the same session anywhere in my application. Since this is a web-applications this would usually mean the same (as one view is executed).

But where is the difference? What is the disadvantage of using one session all the time against using it for my database stuff until my function is done and then creating a new one the next time I want to talk to my DB?

I get that if I use multiple threads, each one should get their own session. But using scoped_session(), I already make sure that problem doesn’t exist, do I?

Please clarify if any of my assumptions are wrong.

sessionmaker() is a factory, it’s there to encourage placing configuration options for creating new Session objects in just one place. It is optional, in that you could just as easily call Session(bind=engine, expire_on_commit=False) anytime you needed a new Session, except that its verbose and redundant, and I wanted to stop the proliferation of small-scale “helpers” that each approached the issue of this redundancy in some new and more confusing way.

So sessionmaker() is just a tool to help you create Session objects when you need them.

Next part. I think the question is, what’s the difference between making a new Session() at various points versus just using one all the way through. The answer, not very much. Session is a container for all the objects you put into it, and then it also keeps track of an open transaction. At the moment you call rollback() or commit(), the transaction is over, and the Session has no connection to the database until it is called upon to emit SQL again. The links it holds to your mapped objects are weak referencing, provided the objects are clean of pending changes, so even in that regard the Session will empty itself out back to a brand new state when your application loses all references to mapped objects. If you leave it with its default "expire_on_commit" setting, then all the objects are expired after a commit. If that Session hangs around for five or twenty minutes, and all kinds of things have changed in the database the next time you use it, it will load all brand new state the next time you access those objects even though they’ve been sitting in memory for twenty minutes.

In web applications, we usually say, hey why don’t you make a brand new Session on each request, rather than using the same one over and over again. This practice ensures that the new request begins “clean”. If some objects from the previous request haven’t been garbage collected yet, and if maybe you’ve turned off "expire_on_commit", maybe some state from the previous request is still hanging around, and that state might even be pretty old. If you’re careful to leave expire_on_commit turned on and to definitely call commit() or rollback() at request end, then it’s fine, but if you start with a brand new Session, then there’s not even any question that you’re starting clean. So the idea to start each request with a new Session is really just the simplest way to make sure you’re starting fresh, and to make the usage of expire_on_commit pretty much optional, as this flag can incur a lot of extra SQL for an operation that calls commit() in the middle of a series of operations. Not sure if this answers your question.

The next round is what you mention about threading. If your app is multithreaded, we recommend making sure the Session in use is local to…something. scoped_session() by default makes it local to the current thread. In a web app, local to the request is in fact even better. Flask-SQLAlchemy actually sends a custom “scope function” to scoped_session() so that you get a request-scoped session. The average Pyramid application sticks the Session into the “request” registry. When using schemes like these, the “create new Session on request start” idea continues to look like the most straightforward way to keep things straight.

In addition to the excellent zzzeek’s answer, here’s a simple recipe to quickly create throwaway, self-enclosed sessions:

from contextlib import contextmanager

from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker

@contextmanager
def db_session(db_url):
    """ Creates a context with an open SQLAlchemy session.
    """
    engine = create_engine(db_url, convert_unicode=True)
    connection = engine.connect()
    db_session = scoped_session(sessionmaker(autocommit=False, autoflush=True, bind=engine))
    yield db_session
    db_session.close()
    connection.close()

Usage:

from mymodels import Foo

with db_session("sqlite://") as db:
    foos = db.query(Foo).all()

I’d really like to be able to print out valid SQL for my application, including values, rather than bind parameters, but it’s not obvious how to do this in SQLAlchemy (by design, I’m fairly sure).

Has anyone solved this problem in a general way?

In the vast majority of cases, the “stringification” of a SQLAlchemy statement or query is as simple as:

print(str(statement))

This applies both to an ORM Query as well as any select() or other statement.

Note: the following detailed answer is being maintained on the sqlalchemy documentation.

To get the statement as compiled to a specific dialect or engine, if the statement itself is not already bound to one you can pass this in to compile():

print(statement.compile(someengine))

or without an engine:

from sqlalchemy.dialects import postgresql
print(statement.compile(dialect=postgresql.dialect()))

When given an ORM Query object, in order to get at the compile() method we only need access the .statement accessor first:

statement = query.statement
print(statement.compile(someengine))

with regards to the original stipulation that bound parameters are to be “inlined” into the final string, the challenge here is that SQLAlchemy normally is not tasked with this, as this is handled appropriately by the Python DBAPI, not to mention bypassing bound parameters is probably the most widely exploited security holes in modern web applications. SQLAlchemy has limited ability to do this stringification in certain circumstances such as that of emitting DDL. In order to access this functionality one can use the ‘literal_binds’ flag, passed to compile_kwargs:

from sqlalchemy.sql import table, column, select

t = table('t', column('x'))

s = select([t]).where(t.c.x == 5)

print(s.compile(compile_kwargs={"literal_binds": True}))

the above approach has the caveats that it is only supported for basic types, such as ints and strings, and furthermore if a bindparam without a pre-set value is used directly, it won’t be able to stringify that either.

To support inline literal rendering for types not supported, implement a TypeDecorator for the target type which includes a TypeDecorator.process_literal_param method:

from sqlalchemy import TypeDecorator, Integer


class MyFancyType(TypeDecorator):
    impl = Integer

    def process_literal_param(self, value, dialect):
        return "my_fancy_formatting(%s)" % value

from sqlalchemy import Table, Column, MetaData

tab = Table('mytable', MetaData(), Column('x', MyFancyType()))

print(
    tab.select().where(tab.c.x > 5).compile(
        compile_kwargs={"literal_binds": True})
)

producing output like:

SELECT mytable.x
FROM mytable
WHERE mytable.x > my_fancy_formatting(5)

This works in python 2 and 3 and is a bit cleaner than before, but requires SA>=1.0.

from sqlalchemy.engine.default import DefaultDialect
from sqlalchemy.sql.sqltypes import String, DateTime, NullType

# python2/3 compatible.
PY3 = str is not bytes
text = str if PY3 else unicode
int_type = int if PY3 else (int, long)
str_type = str if PY3 else (str, unicode)


class StringLiteral(String):
    """Teach SA how to literalize various things."""
    def literal_processor(self, dialect):
        super_processor = super(StringLiteral, self).literal_processor(dialect)

        def process(value):
            if isinstance(value, int_type):
                return text(value)
            if not isinstance(value, str_type):
                value = text(value)
            result = super_processor(value)
            if isinstance(result, bytes):
                result = result.decode(dialect.encoding)
            return result
        return process


class LiteralDialect(DefaultDialect):
    colspecs = {
        # prevent various encoding explosions
        String: StringLiteral,
        # teach SA about how to literalize a datetime
        DateTime: StringLiteral,
        # don't format py2 long integers to NULL
        NullType: StringLiteral,
    }


def literalquery(statement):
    """NOTE: This is entirely insecure. DO NOT execute the resulting strings."""
    import sqlalchemy.orm
    if isinstance(statement, sqlalchemy.orm.Query):
        statement = statement.statement
    return statement.compile(
        dialect=LiteralDialect(),
        compile_kwargs={'literal_binds': True},
    ).string

Demo:

# coding: UTF-8
from datetime import datetime
from decimal import Decimal

from literalquery import literalquery


def test():
    from sqlalchemy.sql import table, column, select

    mytable = table('mytable', column('mycol'))
    values = (
        5,
        u'snowman: ☃',
        b'UTF-8 snowman: \xe2\x98\x83',
        datetime.now(),
        Decimal('3.14159'),
        10 ** 20,  # a long integer
    )

    statement = select([mytable]).where(mytable.c.mycol.in_(values)).limit(1)
    print(literalquery(statement))


if __name__ == '__main__':
    test()

Gives this output: (tested in python 2.7 and 3.4)

SELECT mytable.mycol
FROM mytable
WHERE mytable.mycol IN (5, 'snowman: ☃', 'UTF-8 snowman: ☃',
      '2015-06-24 18:09:29.042517', 3.14159, 100000000000000000000)
 LIMIT 1

Given that what you want makes sense only when debugging, you could start SQLAlchemy with echo=True, to log all SQL queries. For example:

engine = create_engine(
    "mysql://scott:tiger@hostname/dbname",
    encoding="latin1",
    echo=True,
)

This can also be modified for just a single request:

echo=False – if True, the Engine will log all statements as well as a repr() of their parameter lists to the engines logger, which defaults to sys.stdout. The echo attribute of Engine can be modified at any time to turn logging on and off. If set to the string "debug", result rows will be printed to the standard output as well. This flag ultimately controls a Python logger; see Configuring Logging for information on how to configure logging directly.

Source: SQLAlchemy Engine Configuration

If used with Flask, you can simply set

app.config["SQLALCHEMY_ECHO"] = True

to get the same behaviour.

We can use compile method for this purpose. From the docs:

from sqlalchemy.sql import text
from sqlalchemy.dialects import postgresql

stmt = text("SELECT * FROM users WHERE users.name BETWEEN :x AND :y")
stmt = stmt.bindparams(x="m", y="z")

print(stmt.compile(dialect=postgresql.dialect(),compile_kwargs={"literal_binds": True}))

Result:

SELECT * FROM users WHERE users.name BETWEEN 'm' AND 'z'

Warning from docs:

Never use this technique with string content received from untrusted input, such as from web forms or other user-input applications. SQLAlchemy’s facilities to coerce Python values into direct SQL string values are not secure against untrusted input and do not validate the type of data being passed. Always use bound parameters when programmatically invoking non-DDL SQL statements against a relational database.

So building on @zzzeek’s comments on @bukzor’s code I came up with this to easily get a “pretty-printable” query:

def prettyprintable(statement, dialect=None, reindent=True):
    """Generate an SQL expression string with bound parameters rendered inline
    for the given SQLAlchemy statement. The function can also receive a
    `sqlalchemy.orm.Query` object instead of statement.
    can 

    WARNING: Should only be used for debugging. Inlining parameters is not
             safe when handling user created data.
    """
    import sqlparse
    import sqlalchemy.orm
    if isinstance(statement, sqlalchemy.orm.Query):
        if dialect is None:
            dialect = statement.session.get_bind().dialect
        statement = statement.statement
    compiled = statement.compile(dialect=dialect,
                                 compile_kwargs={'literal_binds': True})
    return sqlparse.format(str(compiled), reindent=reindent)

I personally have a hard time reading code which is not indented so I’ve used sqlparse to reindent the SQL. It can be installed with pip install sqlparse.

This code is based on brilliant existing answer from @bukzor. I just added custom render for datetime.datetime type into Oracle’s TO_DATE().

Feel free to update code to suit your database:

import decimal
import datetime

def printquery(statement, bind=None):
    """
    print a query, with values filled in
    for debugging purposes *only*
    for security, you should always separate queries from their values
    please also note that this function is quite slow
    """
    import sqlalchemy.orm
    if isinstance(statement, sqlalchemy.orm.Query):
        if bind is None:
            bind = statement.session.get_bind(
                    statement._mapper_zero_or_none()
            )
        statement = statement.statement
    elif bind is None:
        bind = statement.bind 

    dialect = bind.dialect
    compiler = statement._compiler(dialect)
    class LiteralCompiler(compiler.__class__):
        def visit_bindparam(
                self, bindparam, within_columns_clause=False, 
                literal_binds=False, **kwargs
        ):
            return super(LiteralCompiler, self).render_literal_bindparam(
                    bindparam, within_columns_clause=within_columns_clause,
                    literal_binds=literal_binds, **kwargs
            )
        def render_literal_value(self, value, type_):
            """Render the value of a bind parameter as a quoted literal.

            This is used for statement sections that do not accept bind paramters
            on the target driver/database.

            This should be implemented by subclasses using the quoting services
            of the DBAPI.

            """
            if isinstance(value, basestring):
                value = value.replace("'", "''")
                return "'%s'" % value
            elif value is None:
                return "NULL"
            elif isinstance(value, (float, int, long)):
                return repr(value)
            elif isinstance(value, decimal.Decimal):
                return str(value)
            elif isinstance(value, datetime.datetime):
                return "TO_DATE('%s','YYYY-MM-DD HH24:MI:SS')" % value.strftime("%Y-%m-%d %H:%M:%S")

            else:
                raise NotImplementedError(
                            "Don't know how to literal-quote value %r" % value)            

    compiler = LiteralCompiler(dialect, statement)
    print compiler.process(statement)

I would like to point out that the solutions given above do not “just work” with non-trivial queries. One issue I came across were more complicated types, such as pgsql ARRAYs causing issues. I did find a solution that for me, did just work even with pgsql ARRAYs:

borrowed from: https://gist.github.com/gsakkis/4572159

The linked code seems to be based on an older version of SQLAlchemy. You’ll get an error saying that the attribute _mapper_zero_or_none doesn’t exist. Here’s an updated version that will work with a newer version, you simply replace _mapper_zero_or_none with bind. Additionally, this has support for pgsql arrays:

# adapted from:
# https://gist.github.com/gsakkis/4572159
from datetime import date, timedelta
from datetime import datetime

from sqlalchemy.orm import Query


try:
    basestring
except NameError:
    basestring = str


def render_query(statement, dialect=None):
    """
    Generate an SQL expression string with bound parameters rendered inline
    for the given SQLAlchemy statement.
    WARNING: This method of escaping is insecure, incomplete, and for debugging
    purposes only. Executing SQL statements with inline-rendered user values is
    extremely insecure.
    Based on http://stackoverflow.com/questions/5631078/sqlalchemy-print-the-actual-query
    """
    if isinstance(statement, Query):
        if dialect is None:
            dialect = statement.session.bind.dialect
        statement = statement.statement
    elif dialect is None:
        dialect = statement.bind.dialect

    class LiteralCompiler(dialect.statement_compiler):

        def visit_bindparam(self, bindparam, within_columns_clause=False,
                            literal_binds=False, **kwargs):
            return self.render_literal_value(bindparam.value, bindparam.type)

        def render_array_value(self, val, item_type):
            if isinstance(val, list):
                return "{%s}" % ",".join([self.render_array_value(x, item_type) for x in val])
            return self.render_literal_value(val, item_type)

        def render_literal_value(self, value, type_):
            if isinstance(value, long):
                return str(value)
            elif isinstance(value, (basestring, date, datetime, timedelta)):
                return "'%s'" % str(value).replace("'", "''")
            elif isinstance(value, list):
                return "'{%s}'" % (",".join([self.render_array_value(x, type_.item_type) for x in value]))
            return super(LiteralCompiler, self).render_literal_value(value, type_)

    return LiteralCompiler(dialect, statement).process(statement)

Tested to two levels of nested arrays.