Category: SQLAlchemy

If you are using SQLAlchemy then no doubt you’ve ran into some questions. Find below some more common usages.

Installation:

pip install flask-sqlalchemy && pip freeze > requirements.txt

Table Definition:
You will notice that we are creating a unique constraint, primary keys, foreign keys.

from sqlalchemy.orm import relationship
from sqlalchemy.dialects.postgresql import BOOLEAN, BIT, DATE, JSON, DOUBLE_PRECISION, UUID
from sqlalchemy.sql.functions import current_timestamp
from sqlalchemy.schema import UniqueConstraint

class MyTable(db.Model):
    __tablename__ = 'mytable'
    __table_args__ =  (UniqueConstraint('column_name'),)
    
    primarykey_id = db.Column(db.Integer, primary_key=True, autoincrement="True")
    bit_column = db.Column(BIT)
    text_column = db.Column(db.Text)
    foreignkey_id = db.Column(db.Integer, db.ForeignKey('table.column'))
    foreignkey = relationship("table", foreign_keys=[foreignkey_id ])
    bool_column = db.Column(BOOLEAN)
    created_date = db.Column(db.DateTime, default=current_timestamp())
    guid_column = db.Column(UUID)
    bigint_column = db.Column(db.BigInteger)
    double_precision_column = db.Column(DOUBLE_PRECISION)
    json_column = db.Column(JSON)
    string_column = db.Column(db.String(200))
    
    def __init__(self, primarykey_id, bit_column):
        self.primarykey_id = primarykey_id
        self.bit_column = bit_column

    def __repr__(self):
        return '<MyTable primarykey_id %r Name %r bit_column %r>' % (self.primarykey_id, self.bit_column)

Database Connections:
The default connection is the “SQLALCHEMY_DATABASE_URI”. However you can attach to other DB by using the BINDS as demonstrated below.

SQLALCHEMY_DATABASE_URI = "postgresql://##USER##:##PASS##@##SERVER##/##DB##"

SQLALCHEMY_BINDS = {
	"NAME": "postgresql://##USER##:##PASS##@##SERVER##/##DB##",
}

b.session().execute("", params=dict(), bind=db.get_engine(app, 'NAME'))

Concatenation:

 from sqlalchemy.sql.expression import func

func.concat(Table.Column,'-',Table.Column2)

Case Statements:

 from sqlalchemy.sql.expression import case

case([(column1 == 'some_val', 1)], else_=0)

Ordering:
The example below does multiple columns but you can do just one if need be.

 order_by(Table.Column1.desc(), Table.Column2.asc())

Literals:
If you want to add NULL as a column output you must use literals. You could do this with a numerical data or string or whatever.

 from sqlalchemy import literal

literal(None).label('column_name')

Retrieving Records:

 .first()
.one()
.fetchone()

.all()
.fetchall()

Joins:

 join(Table2, and_(Table2.column_id==Table1.column_id))
outerjoin(Table2, and_(Table2.column_id==Table1.column_id))

Union:

 query1.union(query2)

Select Fields:
This just selects one column but you can comma separate it.

 .with_entities(Table.Column)

Where:
You don’t have to use and_ or or_. Use them as needed.

 from sqlalchemy.sql.expression import and_, or_

.filter(and_(Table.Column1, Table.Column2, or_(Table.Column3_, Table.Column4))))

Not:
Select data where value is not NULL

 Table.Column.isnot(None)

Coalesce:

 from sqlalchemy.sql.functions import  coalesce

coalesce(Table.Column, 'Unknown')

Sum:

 from sqlalchemy.sql.expression import func

func.sum(Table.Column)

Count:

 from sqlalchemy.sql.expression import func

func.count()
func.count('*')

Max:

 from sqlalchemy.sql.expression import func

func.max(Table.Column)

AVG:

 from sqlalchemy.sql.expression import func

func.avg(Table.Column)

Substring:

 from sqlalchemy.sql.expression import func

func.substring(Table.Column, REGEX)

Like:

 Table.Column.like("value")

Not Like:

 from sqlalchemy.sql.expression import not_

not_(Table.Column.like("value"))

Commit & Rollback:

 #commit
session.commit()

#rollback
session.rollback()

Inline SQL:

 .execute('SELECT * FROM SP(:param)', params=dict(param=1))

Cast:

 from sqlalchemy.sql.functions import Cast

Cast(Table.Column, BIT)

Array with Distinct:

 from sqlalchemy.sql.expression import distinct, func

func.array_agg(distinct(Table.Column)).label('column_name')