<s> import datetime <EOL> def max_partition ( <EOL> table , schema = "<STR_LIT:default>" , field = None , filter = None , <EOL> metastore_conn_id = '<STR_LIT>' ) : <EOL> '''<STR_LIT>''' <EOL> from airflow . hooks import HiveMetastoreHook <EOL> if '<STR_LIT:.>' in table : <EOL> schema , table = table . split ( '<STR_LIT:.>' ) <EOL> hh = HiveMetastoreHook ( metastore_conn_id = metastore_conn_id ) <EOL> return hh . max_partition ( <EOL> schema = schema , table_name = table , field = field , filter = filter ) <EOL> def _closest_date ( target_dt , date_list , before_target = None ) : <EOL> '''<STR_LIT>''' <EOL> fb = lambda d : d - target_dt if d >= target_dt else datetime . timedelta . max <EOL> fa = lambda d : d - target_dt if d <= target_dt else datetime . timedelta . min <EOL> fnone = lambda d : target_dt - d if d < target_dt else d - target_dt <EOL> if before_target is None : <EOL> return min ( date_list , key = fnone ) . date ( ) <EOL> if before_target : <EOL> return min ( date_list , key = fb ) . date ( ) <EOL> else : <EOL> return min ( date_list , key = fa ) . date ( ) <EOL> def closest_ds_partition ( <EOL> table , ds , before = True , schema = "<STR_LIT:default>" , <EOL> metastore_conn_id = '<STR_LIT>' ) : <EOL> '''<STR_LIT>''' <EOL> from airflow . hooks import HiveMetastoreHook <EOL> if '<STR_LIT:.>' in table : <EOL> schema , table = table . split ( '<STR_LIT:.>' ) <EOL> hh = HiveMetastoreHook ( metastore_conn_id = metastore_conn_id ) <EOL> partitions = hh . get_partitions ( schema = schema , table_name = table ) <EOL> if not partitions : <EOL> return None <EOL> part_vals = [ list ( p . values ( ) ) [ <NUM_LIT:0> ] for p in partitions ] <EOL> if ds in part_vals : <EOL> return ds <EOL> else : <EOL> parts = [ datetime . datetime . strptime ( pv , '<STR_LIT>' ) <EOL> for pv in part_vals ] <EOL> target_dt = datetime . datetime . strptime ( ds , '<STR_LIT>' ) <EOL> closest_ds = _closest_date ( target_dt , parts , before_target = before ) <EOL> return closest_ds . isoformat ( ) </s>
