'Retrieve an XCom value, optionally meeting certain criteria'

@classmethod
@provide_session
def get_one(cls, execution_date, key=None, task_id=None, dag_id=None, include_prior_dates=False, session=None):

    filters = []
    if key:
        filters.append((cls.key == key))
    if task_id:
        filters.append((cls.task_id == task_id))
    if dag_id:
        filters.append((cls.dag_id == dag_id))
    if include_prior_dates:
        filters.append((cls.execution_date <= execution_date))
    else:
        filters.append((cls.execution_date == execution_date))
    query = session.query(cls.value).filter(and_(*filters)).order_by(cls.execution_date.desc(), cls.timestamp.desc()).limit(1)
    result = query.first()
    if result:
        return result.value
