Python: Flask SQLAlchemy

(Last Updated On: )

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

Installation:

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

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

  1. from sqlalchemy.orm import relationship
  2. from sqlalchemy.dialects.postgresql import BOOLEAN, BIT, DATE, JSON, DOUBLE_PRECISION, UUID
  3. from sqlalchemy.sql.functions import current_timestamp
  4. from sqlalchemy.schema import UniqueConstraint
  5.  
  6. class MyTable(db.Model):
  7.     __tablename__ = 'mytable'
  8.     __table_args__ =  (UniqueConstraint('column_name'),)
  9.     
  10.     primarykey_id = db.Column(db.Integer, primary_key=True, autoincrement="True")
  11.     bit_column = db.Column(BIT)
  12.     text_column = db.Column(db.Text)
  13.     foreignkey_id = db.Column(db.Integer, db.ForeignKey('table.column'))
  14.     foreignkey = relationship("table", foreign_keys=[foreignkey_id ])
  15.     bool_column = db.Column(BOOLEAN)
  16.     created_date = db.Column(db.DateTime, default=current_timestamp())
  17.     guid_column = db.Column(UUID)
  18.     bigint_column = db.Column(db.BigInteger)
  19.     double_precision_column = db.Column(DOUBLE_PRECISION)
  20.     json_column = db.Column(JSON)
  21.     string_column = db.Column(db.String(200))
  22.     
  23.     def __init__(self, primarykey_id, bit_column):
  24.         self.primarykey_id = primarykey_id
  25.         self.bit_column = bit_column
  26.  
  27.     def __repr__(self):
  28.         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.

  1. SQLALCHEMY_DATABASE_URI = "postgresql://##USER##:##PASS##@##SERVER##/##DB##"
  2.  
  3. SQLALCHEMY_BINDS = {
  4. 	"NAME": "postgresql://##USER##:##PASS##@##SERVER##/##DB##",
  5. }
  6.  
  7. b.session().execute("", params=dict(), bind=db.get_engine(app, 'NAME'))

Concatenation:

  1.  from sqlalchemy.sql.expression import func
  2.  
  3. func.concat(Table.Column,'-',Table.Column2)

Case Statements:

  1.  from sqlalchemy.sql.expression import case
  2.  
  3. case([(column1 == 'some_val', 1)], else_=0)

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

  1.  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.

  1.  from sqlalchemy import literal
  2.  
  3. literal(None).label('column_name')

Retrieving Records:

  1.  .first()
  2. .one()
  3. .fetchone()
  4.  
  5. .all()
  6. .fetchall()

Joins:

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

Union:

  1.  query1.union(query2)

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

  1.  .with_entities(Table.Column)

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

  1.  from sqlalchemy.sql.expression import and_, or_
  2.  
  3. .filter(and_(Table.Column1, Table.Column2, or_(Table.Column3_, Table.Column4))))

Not:
Select data where value is not NULL

  1.  Table.Column.isnot(None)

Coalesce:

  1.  from sqlalchemy.sql.functions import  coalesce
  2.  
  3. coalesce(Table.Column, 'Unknown')

Sum:

  1.  from sqlalchemy.sql.expression import func
  2.  
  3. func.sum(Table.Column)

Count:

  1.  from sqlalchemy.sql.expression import func
  2.  
  3. func.count()
  4. func.count('*')

Max:

  1.  from sqlalchemy.sql.expression import func
  2.  
  3. func.max(Table.Column)

AVG:

  1.  from sqlalchemy.sql.expression import func
  2.  
  3. func.avg(Table.Column)

Substring:

  1.  from sqlalchemy.sql.expression import func
  2.  
  3. func.substring(Table.Column, REGEX)

Like:

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

Not Like:

  1.  from sqlalchemy.sql.expression import not_
  2.  
  3. not_(Table.Column.like("value"))

Commit & Rollback:

  1.  #commit
  2. session.commit()
  3.  
  4. #rollback
  5. session.rollback()

Inline SQL:

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

Cast:

  1.  from sqlalchemy.sql.functions import Cast
  2.  
  3. Cast(Table.Column, BIT)

Array with Distinct:

  1.  from sqlalchemy.sql.expression import distinct, func
  2.  
  3. func.array_agg(distinct(Table.Column)).label('column_name')