Skip to content

Embed URL

HTTPS clone URL

Subversion checkout URL

You can clone with
or
.
Download ZIP
UUID_SHORT default for MySQL on SQLAlchemy
#!/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()))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Something went wrong with that request. Please try again.