UUID_SHORT default for MySQL on SQLAlchemy

uuid_short_primary_key_sa.py

#!/usr/bin/env python3

from sqlalchemy import BigInteger, Column, create_engine, DDL, event
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.schema import FetchedValue
from sqlalchemy.sql.expression import func

Base = declarative_base()

class PatientOutputMixin(object):
    '''
    Mixin to output human readable representations of models.
    '''
    def __str__(self):
        return '{}'.format(self.patient_id)

    def __repr__(self):
        return str(self)


class Patient1(Base, PatientOutputMixin):
    '''
    First version of ``Patient`` model.
    '''
    __tablename__ = 'patient_1'

    patient_id = Column(BigInteger, primary_key=True,
                        default=func.uuid_short())

# the following trigger is only required if columns are inserted in the table
# not using the above model/table definition, otherwise it is redundant
create_before_insert_trigger = DDL('''
CREATE TRIGGER before_insert_%(table)s BEFORE INSERT ON %(table)s
FOR EACH ROW BEGIN
    IF NEW.patient_id IS NULL THEN
        SET NEW.patient_id = UUID_SHORT();
    END IF;
END
''')

event.listen(Patient1.__table__, 'after_create',
             create_before_insert_trigger.execute_if(dialect='mysql'))
# end of optional trigger definition


class Patient2(Base, PatientOutputMixin):
    '''
    Second version of ``Patient`` model.
    '''
    __tablename__ = 'patient_2'

    patient_id = Column(BigInteger, primary_key=True,
                        default=0, server_default=FetchedValue())

create_before_insert_trigger = DDL('''
CREATE TRIGGER before_insert_%(table)s BEFORE INSERT ON %(table)s
FOR EACH ROW BEGIN
    SET NEW.patient_id = UUID_SHORT();
END
''')

event.listen(Patient2.__table__, 'after_create',
             create_before_insert_trigger.execute_if(dialect='mysql'))


# test models

engine = create_engine('mysql+oursql://test:test@localhost/test?charset=utf8')

Base.metadata.bind = engine
Base.metadata.drop_all()
Base.metadata.create_all()

Session = sessionmaker(bind=engine)

session = Session()

for patient_model in [Patient1, Patient2]:
    session.add(patient_model())
    session.add(patient_model())
    session.commit()

    print('{} instances: {}'.format(patient_model.__name__,
                                    session.query(patient_model).all()))