public
Created

UUID_SHORT default for MySQL on SQLAlchemy

  • Download Gist
uuid_short_primary_key_sa.py
Python
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85
#!/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()))

Please sign in to comment on this gist.

Something went wrong with that request. Please try again.