(
        self,
        source: BaseRelation,
        loaded_at_field: str,
        filter: Optional[str],
        manifest: Optional[Manifest] = None
    )
return {
            'max_loaded_at': max_loaded_at,
            'snapshotted_at': snapshotted_at,
            'age': age,
        }
Calculate the freshness of sources in dbt, and return it
Calculate the freshness of sources in dbt, and return it
Calculate the freshness of sources in dbt and return it
def calculate_freshness(
        self,
        source: BaseRelation,
        loaded_at_field: str,
        filter: Optional[str],
        manifest: Optional[Manifest] = None
    ) -> Dict[str, Any]:
        """Calculate the freshness of sources in dbt, and return it"""
        kwargs: Dict[str, Any] = {
            'source': source,
            'loaded_at_field': loaded_at_field,
            'filter': filter,
        }

        # run the macro
        table = self.execute_macro(
            FRESHNESS_MACRO_NAME,
            kwargs=kwargs,
            manifest=manifest
        )
        # now we have a 1-row table of the maximum `loaded_at_field` value and
        # the current time according to the db.
        if len(table) != 1 or len(table[0]) != 2:
            raise_compiler_error(
                'Got an invalid result from "{}" macro: {}'.format(
                    FRESHNESS_MACRO_NAME, [tuple(r) for r in table]
                )
            )
        if table[0][0] is None:
            # no records in the table, so really the max_loaded_at was
            # infinitely long ago. Just call it 0:00 January 1 year UTC
            max_loaded_at = datetime(1, 1, 1, 0, 0, 0, tzinfo=pytz.UTC)
        else:
            max_loaded_at = _utc(table[0][0], source, loaded_at_field)

        snapshotted_at = _utc(table[0][1], source, loaded_at_field)
        age = (snapshotted_at - max_loaded_at).total_seconds()
        return {
            'max_loaded_at': max_loaded_at,
            'snapshotted_at': snapshotted_at,
            'age': age,
        }
def calculate_freshness ( self , source : BaseRelation , loaded_at_field : str , filter : Optional [ str ] , manifest : Optional [ Manifest ] = None ) -> Dict [ str , Any ] : kwargs : Dict [ str , Any ] = { 'source' : source , 'loaded_at_field' : loaded_at_field , 'filter' : filter , } # run the macro table = self . execute_macro ( FRESHNESS_MACRO_NAME , kwargs = kwargs , manifest = manifest ) # now we have a 1-row table of the maximum `loaded_at_field` value and # the current time according to the db. if len ( table ) != 1 or len ( table [ 0 ] ) != 2 : raise_compiler_error ( 'Got an invalid result from "{}" macro: {}' . format ( FRESHNESS_MACRO_NAME , [ tuple ( r ) for r in table ] ) ) if table [ 0 ] [ 0 ] is None : # no records in the table, so really the max_loaded_at was # infinitely long ago. Just call it 0:00 January 1 year UTC max_loaded_at = datetime ( 1 , 1 , 1 , 0 , 0 , 0 , tzinfo = pytz . UTC ) else : max_loaded_at = _utc ( table [ 0 ] [ 0 ] , source , loaded_at_field ) snapshotted_at = _utc ( table [ 0 ] [ 1 ] , source , loaded_at_field ) age = ( snapshotted_at - max_loaded_at ) . total_seconds ( ) return { 'max_loaded_at' : max_loaded_at , 'snapshotted_at' : snapshotted_at , 'age' : age , }