All notes
Sqlalchemy

Examples


import sqlalchemy

try:
    URL = "mssql+pymssql://%s:%[email protected]%s/%s" % (cfg['dbUser'], cfg['dbPass'], cfg['dbHost'], cfg['dbName'])
    engine = sqlalchemy.create_engine(URL, encoding='utf-8', max_overflow=3, pool_size=5, pool_recycle=3600, pool_timeout=30)
except Exception:
    logging.error(traceback.format_exc())

# "sqlStr" here is not safe. See more below for safer way.
results = engine.execute(sqlStr)
    
output = []
for row in results:
    tmp = {}
    for i, j in enumerate(row):
        tmp[apiKeys[i]] = j
    output.append(tmp)

# or

output = [{apiKeys[i]: j} for i,j in enumerate(row)]

with engine.connect() as conn:
    conn.execute(table.insert(),
        {"id":1, "value":"v1"},
        {"id":2, "value":"v2"}
    )

Engines

create_engine


engine = create_engine("mysql://scott:[email protected]/dbname", encoding='latin1', echo=True)

Close, disposal

SQLAlchemyDocs: engine disposal.

The Engine is intended to normally be a permanent fixture established up-front and maintained throughout the lifespan of an application. It is not intended to be created and disposed on a per-connection basis.

The Engine can be explicitly disposed using the Engine.dispose() method. This disposes of the engine's underlying connection pool and replaces it with a new one that's empty. Valid use cases for calling Engine.dispose() include:

Connections that are checked out are not discarded when the engine is disposed or garbage collected, as these connections are still strongly referenced elsewhere by the application.

Execute


execute(object, *multiparams, **params)

conn.execute(table.insert(),
    {"id":1, "value":"v1"},
    {"id":2, "value":"v2"}
)

# or individual key/values interpreted by **params:

conn.execute(
    table.insert(), id=1, value="v1"
)

# In the case that a plain SQL string is passed, and the underlying DBAPI accepts positional bind parameters, a collection of tuples or individual values in *multiparams may be passed:
conn.execute(
    "INSERT INTO table (id, value) VALUES (?, ?)",
    (1, "v1"), (2, "v2")
)
conn.execute(
    "INSERT INTO table (id, value) VALUES (?, ?)",
    1, "v1"
)

text

SqlAlchemy: sql.expression.text.


from sqlalchemy import text

t = text("SELECT * FROM users")
result = connection.execute(t)

# Bind parameters are specified by name, using the format ":name"
t = text("SELECT * FROM users WHERE id=:user_id")
result = connection.execute(t, user_id=12)
# Use \ to escape:
t = text("SELECT * FROM users WHERE name='\:username'")

ORM


from sqlalchemy import create_engine
# Test in memory
engine = create_engine('sqlite:///:memory:', echo=True)
# engine = create_engine('sqlite:///test.sqlite', echo=True)

from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

from sqlalchemy import Column, Integer, String
class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    # on SQLite and Postgresql, "varchar" is valid, so we don't specify length (such as "String(50)") here.
    name = Column(String)
    fullname = Column(String)
    password = Column(String)

    def __repr__(self):
       return "User(name='%s', fullname='%s', password='%s')" % (self.name, self.fullname, self.password)

print(User.__table__.__class__)
# class 'sqlalchemy.sql.schema.Table'

Base.metadata.create_all(engine)
# Special commands are first emitted to check for the presence of the users table, and following that the actual CREATE TABLE statement:
# SELECT CAST('test plain returns' AS VARCHAR(60)) AS anon_1
# SELECT CAST('test unicode returns' AS VARCHAR(60)) AS anon_1
# PRAGMA table_info("users")
# CREATE TABLE users (
#    id INTEGER NOT NULL,
#       name VARCHAR,
#       fullname VARCHAR,
#       password VARCHAR,
#       PRIMARY KEY (id)
# )
# COMMIT

ed_user = User(name='ed', fullname='Ed Jones', password='edspassword')

from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine)

# Or
# Session = sessionmaker()
# Session.configure(bind=engine)

session = Session()
session.add(ed_user) # the instance is pending.

our_user = session.query(User).filter_by(name='ed').first()
ed_user is our_user
# True

session.add_all([
	User(name='wendy', fullname='Wendy Williams', password='foobar'),
	User(name='mary', fullname='Mary Contrary', password='xxg527'),
	User(name='fred', fullname='Fred Flinstone', password='blah')])

for instance in session.query(User).order_by(User.id):
	print(instance.name, instance.fullname)

for name, fullname in session.query(User.name, User.fullname):
	print(name, fullname)

session.commit()

# Rollback
# session.rollback()

The Session will issue the SQL to persist Ed Jones as soon as is needed, using a process known as a flush. If we query the database for Ed Jones, all pending information will first be flushed, and the query is issued immediately thereafter.


# A full, foolproof table:
class User(Base):
    __tablename__ = 'users'
    # Firebird and Oracle require "sequences" to generate new primary key identifiers.
    id = Column(Integer, Sequence('user_id_seq'), primary_key=True)
    name = Column(String(50))
    fullname = Column(String(50))
    password = Column(String(12))

FAQ

Connect to MySQL


pip search mysql
pip install mysql-connector-python
# Then use "mysql+mysqlconnector"

Connect to SQL Server/ mssql

SQLalchemyDocs: mssql.


import sqlalchemy
engine = sqlalchemy.create_engine("mssql+pymssql://user:[email protected]/dbName")

Output sql

StackOverflow: debugging displaying SQL.

In addition to "echo" parameter of create_engine():


import logging
logging.basicConfig()
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)