<s> import json <EOL> from arbalest . s3 import normalize_path <EOL> from arbalest . sql import Database <EOL> class Manifest ( object ) : <EOL> def __init__ ( self , metadata , source , schema , bucket ) : <EOL> self . metadata = metadata <EOL> self . source = source <EOL> self . schema = schema <EOL> self . bucket = bucket <EOL> self . file_name = '<STR_LIT>' . format ( schema . table ) <EOL> self . journal_file_name = '<STR_LIT>' . format ( schema . table ) <EOL> @ property <EOL> def all_keys ( self ) : <EOL> return [ k . name for k in self . bucket . list ( self . source ) if <EOL> not k . name . endswith ( '<STR_LIT:/>' ) ] <EOL> @ property <EOL> def manifest_key ( self ) : <EOL> return normalize_path ( '<STR_LIT>' . format ( self . metadata , self . file_name ) ) <EOL> @ property <EOL> def journal_key ( self ) : <EOL> return normalize_path ( <EOL> '<STR_LIT>' . format ( self . metadata , self . journal_file_name ) ) <EOL> @ property <EOL> def manifest_url ( self ) : <EOL> return '<STR_LIT>' . format ( self . bucket . name , self . manifest_key ) <EOL> def journal ( self ) : <EOL> journal = self . bucket . get ( self . journal_key ) <EOL> if journal . exists ( ) : <EOL> return json . loads ( journal . get_contents_as_string ( ) ) <EOL> else : <EOL> return [ ] <EOL> def get ( self ) : <EOL> updated_journal = self . all_keys <EOL> journal = self . journal ( ) <EOL> keys = list ( set ( updated_journal ) - set ( journal ) ) <EOL> return { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : [ <EOL> { '<STR_LIT:url>' : '<STR_LIT>' . format ( self . bucket . name , key ) , <EOL> '<STR_LIT>' : True } for key in keys <EOL> ] <EOL> } , <EOL> '<STR_LIT>' : updated_journal <EOL> } <EOL> def save ( self ) : <EOL> manifest = self . get ( ) <EOL> self . bucket . save ( self . manifest_key , json . dumps ( manifest [ '<STR_LIT>' ] ) ) <EOL> return manifest [ '<STR_LIT>' ] <EOL> def commit ( self , saved_keys ) : <EOL> self . bucket . save ( self . journal_key , json . dumps ( saved_keys ) ) <EOL> def exists ( self ) : <EOL> return self . bucket . get ( self . manifest_key ) . exists ( ) <EOL> def journal_exists ( self ) : <EOL> return self . bucket . get ( self . journal_key ) . exists ( ) <EOL> class SqlManifest ( object ) : <EOL> def __init__ ( self , metadata , source , schema , bucket , db_connection ) : <EOL> self . metadata = metadata <EOL> self . source = source <EOL> self . schema = schema <EOL> self . bucket = bucket <EOL> if isinstance ( db_connection , Database ) : <EOL> self . database = db_connection <EOL> else : <EOL> self . database = Database ( db_connection ) <EOL> self . file_name = '<STR_LIT>' . format ( schema . table ) <EOL> self . journal_file_name = '<STR_LIT>' . format ( schema . table ) <EOL> @ property <EOL> def all_keys ( self ) : <EOL> return ( k . name for k in self . bucket . list ( self . source ) if <EOL> not k . name . endswith ( '<STR_LIT:/>' ) ) <EOL> @ property <EOL> def manifest_key ( self ) : <EOL> return normalize_path ( '<STR_LIT>' . format ( self . metadata , self . file_name ) ) <EOL> @ property <EOL> def journal_key ( self ) : <EOL> return normalize_path ( <EOL> '<STR_LIT>' . format ( self . metadata , self . journal_file_name ) ) <EOL> @ property <EOL> def manifest_url ( self ) : <EOL> return '<STR_LIT>' . format ( self . bucket . name , self . manifest_key ) <EOL> def journal ( self ) : <EOL> journal = self . bucket . get ( self . journal_key ) <EOL> if journal . exists ( ) : <EOL> journal . get_contents_to_filename ( self . journal_file_name ) <EOL> self . database . open ( ) <EOL> self . database . execute ( '<STR_LIT>' ) <EOL> return ( row [ <NUM_LIT:0> ] for row in self . database . fetchall ( ) ) <EOL> else : <EOL> self . database . open ( ) <EOL> self . database . execute ( <EOL> '<STR_LIT>' ) <EOL> self . database . commit ( ) <EOL> self . database . close ( ) <EOL> return [ ] <EOL> def get ( self ) : <EOL> updated_journal = self . all_keys <EOL> journal = self . journal ( ) <EOL> keys = list ( set ( updated_journal ) - set ( journal ) ) <EOL> return { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : ( <EOL> { '<STR_LIT:url>' : '<STR_LIT>' . format ( self . bucket . name , key ) , <EOL> '<STR_LIT>' : True } for key in keys <EOL> ) <EOL> } , <EOL> '<STR_LIT>' : updated_journal <EOL> } <EOL> def save ( self ) : <EOL> entries = self . get ( ) [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> offset = <NUM_LIT:0> <EOL> offsets = [ ] <EOL> last_entry = None <EOL> with open ( self . file_name , '<STR_LIT:wb>' ) as f : <EOL> f . seek ( <NUM_LIT:0> ) <EOL> f . truncate ( ) <EOL> offset += self . __write ( f , '<STR_LIT>' ) <EOL> offsets . append ( offset ) <EOL> offset += self . __write ( f , '<STR_LIT>' ) <EOL> offsets . append ( offset ) <EOL> for entry in entries : <EOL> line = json . dumps ( entry ) + '<STR_LIT>' <EOL> offset += self . __write ( f , line ) <EOL> offsets . append ( offset ) <EOL> last_entry = line <EOL> f . seek ( offsets [ - <NUM_LIT:2> ] ) <EOL> f . truncate ( ) <EOL> if last_entry is not None : <EOL> line = last_entry [ <NUM_LIT:0> : - <NUM_LIT:2> ] + '<STR_LIT:\n>' <EOL> f . write ( line ) <EOL> f . write ( '<STR_LIT>' ) <EOL> self . bucket . get ( self . manifest_key ) . set_contents_from_filename ( <EOL> self . file_name ) <EOL> def commit ( self , saved_keys ) : <EOL> self . database . open ( ) <EOL> self . database . execute ( '<STR_LIT>' ) <EOL> for key in saved_keys : <EOL> self . database . execute ( '<STR_LIT>' , ( key , ) ) <EOL> self . database . commit ( ) <EOL> self . database . close ( ) <EOL> self . bucket . get ( self . journal_key ) . set_contents_from_filename ( <EOL> self . journal_file_name ) <EOL> def exists ( self ) : <EOL> return self . bucket . get ( self . manifest_key ) . exists ( ) <EOL> def journal_exists ( self ) : <EOL> return self . bucket . get ( self . journal_key ) . exists ( ) <EOL> @ staticmethod <EOL> def __write ( fd , line ) : <EOL> fd . write ( line ) <EOL> return len ( line ) </s>
