Source code for examples.custom_attributes.active_column_defaults
"""Illustrates use of the :meth:`.AttributeEvents.init_scalar`event, in conjunction with Core column defaults to provideORM objects that automatically produce the default valuewhen an un-set attribute is accessed."""fromsqlalchemyimporteventdefconfigure_listener(mapper,class_):"""Establish attribute setters for every default-holding column on the given mapper."""# iterate through ColumnProperty objectsforcol_attrinmapper.column_attrs:# look at the Column mapped by the ColumnProperty# (we look at the first column in the less common case# of a property mapped to multiple columns at once)column=col_attr.columns[0]# if the Column has a "default", set up a listenerifcolumn.defaultisnotNone:default_listener(col_attr,column.default)defdefault_listener(col_attr,default):"""Establish a default-setting listener. Given a class attribute and a :class:`.DefaultGenerator` instance. The default generator should be a :class:`.ColumnDefault` object with a plain Python value or callable default; otherwise, the appropriate behavior for SQL functions and defaults should be determined here by the user integrating this feature. """@event.listens_for(col_attr,"init_scalar",retval=True,propagate=True)definit_scalar(target,value,dict_):ifdefault.is_callable:# the callable of ColumnDefault always accepts a context# argument; we can pass it as None here.value=default.arg(None)elifdefault.is_scalar:value=default.argelse:# default is a Sequence, a SQL expression, server# side default generator, or other non-Python-evaluable# object. The feature here can't easily support this. This# can be made to return None, rather than raising,# or can procure a connection from an Engine# or Session and actually run the SQL, if desired.raiseNotImplementedError("Can't invoke pre-default for a SQL-level column default")# set the value in the given dict_; this won't emit any further# attribute set events or create attribute "history", but the value# will be used in the INSERT statementdict_[col_attr.key]=value# return the value as wellreturnvalueif__name__=="__main__":fromsqlalchemyimportColumn,Integer,DateTime,create_enginefromsqlalchemy.ormimportSessionfromsqlalchemy.ext.declarativeimportdeclarative_baseimportdatetimeBase=declarative_base()event.listen(Base,"mapper_configured",configure_listener,propagate=True)classWidget(Base):__tablename__="widget"id=Column(Integer,primary_key=True)radius=Column(Integer,default=30)timestamp=Column(DateTime,default=datetime.datetime.now)e=create_engine("sqlite://",echo=True)Base.metadata.create_all(e)w1=Widget()# not persisted at all, default values are present the moment# we access themassertw1.radius==30# this line will invoke the datetime.now() function, and establish# its return value upon the w1 instance, such that the# Column-level default for the "timestamp" column will no longer fire# off.current_time=w1.timestampassertcurrent_time>datetime.datetime.now()-datetime.timedelta(seconds=5)# persistsess=Session(e)sess.add(w1)sess.commit()# data is persisted. The timestamp is also the one we generated above;# e.g. the default wasn't re-invoked later.assertsess.query(Widget.radius,Widget.timestamp).first()==(30,current_time,)