Last active
May 12, 2020 19:36
-
-
Save edelooff/211e5555a2e9b49967d24a08b37cedd5 to your computer and use it in GitHub Desktop.
SQLAlchemy columns with a hybrid property boolean flag to check for presence (non-nullness)
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from sqlalchemy import ( | |
Column, | |
Integer, | |
DateTime, | |
Text, | |
create_engine, | |
func) | |
from sqlalchemy.ext.declarative import declarative_base | |
from sqlalchemy.ext.hybrid import hybrid_property | |
from sqlalchemy.inspection import inspect | |
from sqlalchemy.orm import Session | |
Base = declarative_base() | |
def column_flag(column, **options): | |
def target_key(instance): | |
target = inspect(instance).mapper.get_property_by_column(column) | |
return target.key | |
def fget(self): | |
return getattr(self, target_key(self)) is not None | |
def fset(self, value, default=options.get('default')): | |
if not isinstance(value, bool): | |
raise TypeError('Flag only accepts boolean values') | |
return setattr(self, target_key(self), default if value else None) | |
def expr(cls): | |
return column.isnot(None) | |
return hybrid_property( | |
fget=fget, | |
fset=fset if 'default' in options else None, | |
expr=expr) | |
class Article(Base): | |
__tablename__ = 'article' | |
id = Column(Integer, primary_key=True) | |
content = Column(Text) | |
published_at = Column('publication_date', DateTime) | |
is_published = column_flag(published_at, default=func.now()) | |
def main(): | |
engine = create_engine('sqlite://', echo=True) | |
Base.metadata.create_all(bind=engine) | |
session = Session(bind=engine) | |
art1 = Article(content='First post', published_at=func.now()) | |
art2 = Article(content='Tentative content') | |
session.add_all([art1, art2]) | |
session.flush() | |
count_total = session.query(Article).count() | |
count_published = session.query(Article)\ | |
.filter(Article.is_published)\ | |
.count() | |
print(f'Articles published out of total: {count_published}/{count_total}') | |
assert art1.is_published | |
assert not art2.is_published | |
# Publish article 2 | |
art2.is_published = True | |
session.flush() | |
print(f'Article 2 was published at {art2.published_at}') | |
# Unpublish article 2 | |
art2.is_published = False | |
session.flush() | |
print(f'Article 2 publication date: {art2.published_at}') | |
if __name__ == '__main__': | |
main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment