<s> import logging <EOL> import sqlite3 <EOL> import os <EOL> from twisted . internet import defer <EOL> from twisted . python . failure import Failure <EOL> from twisted . enterprise import adbapi <EOL> from lbrynet . core . Error import DuplicateStreamHashError , NoSuchStreamHashError <EOL> from lbrynet . core . sqlite_helpers import rerun_if_locked <EOL> log = logging . getLogger ( __name__ ) <EOL> class DBLBRYFileMetadataManager ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , db_dir ) : <EOL> self . db_dir = db_dir <EOL> self . stream_info_db = None <EOL> self . stream_blob_db = None <EOL> self . stream_desc_db = None <EOL> def setup ( self ) : <EOL> return self . _open_db ( ) <EOL> def stop ( self ) : <EOL> self . db_conn = None <EOL> return defer . succeed ( True ) <EOL> def get_all_streams ( self ) : <EOL> return self . _get_all_streams ( ) <EOL> def save_stream ( self , stream_hash , file_name , key , suggested_file_name , blobs ) : <EOL> d = self . _store_stream ( stream_hash , file_name , key , suggested_file_name ) <EOL> d . addCallback ( lambda _ : self . add_blobs_to_stream ( stream_hash , blobs ) ) <EOL> return d <EOL> def get_stream_info ( self , stream_hash ) : <EOL> return self . _get_stream_info ( stream_hash ) <EOL> def check_if_stream_exists ( self , stream_hash ) : <EOL> return self . _check_if_stream_exists ( stream_hash ) <EOL> def delete_stream ( self , stream_hash ) : <EOL> return self . _delete_stream ( stream_hash ) <EOL> def add_blobs_to_stream ( self , stream_hash , blobs ) : <EOL> return self . _add_blobs_to_stream ( stream_hash , blobs , ignore_duplicate_error = True ) <EOL> def get_blobs_for_stream ( self , stream_hash , start_blob = None , end_blob = None , count = None , reverse = False ) : <EOL> log . info ( "<STR_LIT>" , str ( count ) ) <EOL> def get_positions_of_start_and_end ( ) : <EOL> if start_blob is not None : <EOL> d1 = self . _get_blob_num_by_hash ( stream_hash , start_blob ) <EOL> else : <EOL> d1 = defer . succeed ( None ) <EOL> if end_blob is not None : <EOL> d2 = self . _get_blob_num_by_hash ( stream_hash , end_blob ) <EOL> else : <EOL> d2 = defer . succeed ( None ) <EOL> dl = defer . DeferredList ( [ d1 , d2 ] ) <EOL> def get_positions ( results ) : <EOL> start_num = None <EOL> end_num = None <EOL> if results [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] is True : <EOL> start_num = results [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] <EOL> if results [ <NUM_LIT:1> ] [ <NUM_LIT:0> ] is True : <EOL> end_num = results [ <NUM_LIT:1> ] [ <NUM_LIT:1> ] <EOL> return start_num , end_num <EOL> dl . addCallback ( get_positions ) <EOL> return dl <EOL> def get_blob_infos ( nums ) : <EOL> start_num , end_num = nums <EOL> return self . _get_further_blob_infos ( stream_hash , start_num , end_num , <EOL> count , reverse ) <EOL> d = get_positions_of_start_and_end ( ) <EOL> d . addCallback ( get_blob_infos ) <EOL> return d <EOL> def get_stream_of_blob ( self , blob_hash ) : <EOL> return self . _get_stream_of_blobhash ( blob_hash ) <EOL> def save_sd_blob_hash_to_stream ( self , stream_hash , sd_blob_hash ) : <EOL> return self . _save_sd_blob_hash_to_stream ( stream_hash , sd_blob_hash ) <EOL> def get_sd_blob_hashes_for_stream ( self , stream_hash ) : <EOL> return self . _get_sd_blob_hashes_for_stream ( stream_hash ) <EOL> def _open_db ( self ) : <EOL> self . db_conn = adbapi . ConnectionPool ( "<STR_LIT>" , ( os . path . join ( self . db_dir , "<STR_LIT>" ) ) , <EOL> check_same_thread = False ) <EOL> def create_tables ( transaction ) : <EOL> transaction . execute ( "<STR_LIT>" + <EOL> "<STR_LIT>" + <EOL> "<STR_LIT>" + <EOL> "<STR_LIT>" + <EOL> "<STR_LIT>" + <EOL> "<STR_LIT:)>" ) <EOL> transaction . execute ( "<STR_LIT>" + <EOL> "<STR_LIT>" + <EOL> "<STR_LIT>" + <EOL> "<STR_LIT>" + <EOL> "<STR_LIT>" + <EOL> "<STR_LIT>" + <EOL> "<STR_LIT>" + <EOL> "<STR_LIT:)>" ) <EOL> transaction . execute ( "<STR_LIT>" + <EOL> "<STR_LIT>" + <EOL> "<STR_LIT>" + <EOL> "<STR_LIT>" + <EOL> "<STR_LIT:)>" ) <EOL> return self . db_conn . runInteraction ( create_tables ) <EOL> @ rerun_if_locked <EOL> def _delete_stream ( self , stream_hash ) : <EOL> d = self . db_conn . runQuery ( "<STR_LIT>" , ( stream_hash , ) ) <EOL> d . addCallback ( lambda result : result [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] if len ( result ) else Failure ( NoSuchStreamHashError ( stream_hash ) ) ) <EOL> def do_delete ( transaction , s_h ) : <EOL> transaction . execute ( "<STR_LIT>" , ( s_h , ) ) <EOL> transaction . execute ( "<STR_LIT>" , ( s_h , ) ) <EOL> transaction . execute ( "<STR_LIT>" , ( s_h , ) ) <EOL> d . addCallback ( lambda s_h : self . db_conn . runInteraction ( do_delete , s_h ) ) <EOL> return d <EOL> @ rerun_if_locked <EOL> def _store_stream ( self , stream_hash , name , key , suggested_file_name ) : <EOL> d = self . db_conn . runQuery ( "<STR_LIT>" , <EOL> ( stream_hash , key , name , suggested_file_name ) ) <EOL> def check_duplicate ( err ) : <EOL> if err . check ( sqlite3 . IntegrityError ) : <EOL> raise DuplicateStreamHashError ( stream_hash ) <EOL> return err <EOL> d . addErrback ( check_duplicate ) <EOL> return d <EOL> @ rerun_if_locked <EOL> def _get_all_streams ( self ) : <EOL> d = self . db_conn . runQuery ( "<STR_LIT>" ) <EOL> d . addCallback ( lambda results : [ r [ <NUM_LIT:0> ] for r in results ] ) <EOL> return d <EOL> @ rerun_if_locked <EOL> def _get_stream_info ( self , stream_hash ) : <EOL> d = self . db_conn . runQuery ( "<STR_LIT>" , <EOL> ( stream_hash , ) ) <EOL> d . addCallback ( lambda result : result [ <NUM_LIT:0> ] if len ( result ) else Failure ( NoSuchStreamHashError ( stream_hash ) ) ) <EOL> return d <EOL> @ rerun_if_locked <EOL> def _check_if_stream_exists ( self , stream_hash ) : <EOL> d = self . db_conn . runQuery ( "<STR_LIT>" , ( stream_hash , ) ) <EOL> d . addCallback ( lambda r : True if len ( r ) else False ) <EOL> return d <EOL> @ rerun_if_locked <EOL> def _get_blob_num_by_hash ( self , stream_hash , blob_hash ) : <EOL> d = self . db_conn . runQuery ( "<STR_LIT>" , <EOL> ( stream_hash , blob_hash ) ) <EOL> d . addCallback ( lambda r : r [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] if len ( r ) else None ) <EOL> return d <EOL> @ rerun_if_locked <EOL> def _get_further_blob_infos ( self , stream_hash , start_num , end_num , count = None , reverse = False ) : <EOL> params = [ ] <EOL> q_string = "<STR_LIT>" <EOL> q_string += "<STR_LIT>" <EOL> q_string += "<STR_LIT>" <EOL> params . append ( stream_hash ) <EOL> if start_num is not None : <EOL> q_string += "<STR_LIT>" <EOL> params . append ( start_num ) <EOL> if end_num is not None : <EOL> q_string += "<STR_LIT>" <EOL> params . append ( end_num ) <EOL> q_string += "<STR_LIT>" <EOL> if reverse is True : <EOL> q_string += "<STR_LIT>" <EOL> if count is not None : <EOL> q_string += "<STR_LIT>" <EOL> params . append ( count ) <EOL> q_string += "<STR_LIT>" <EOL> return self . db_conn . runQuery ( q_string , tuple ( params ) ) <EOL> @ rerun_if_locked <EOL> def _add_blobs_to_stream ( self , stream_hash , blob_infos , ignore_duplicate_error = False ) : <EOL> def add_blobs ( transaction ) : <EOL> for blob_info in blob_infos : <EOL> try : <EOL> transaction . execute ( "<STR_LIT>" , <EOL> ( blob_info . blob_hash , stream_hash , blob_info . blob_num , <EOL> blob_info . iv , blob_info . length ) ) <EOL> except sqlite3 . IntegrityError : <EOL> if ignore_duplicate_error is False : <EOL> raise <EOL> return self . db_conn . runInteraction ( add_blobs ) <EOL> @ rerun_if_locked <EOL> def _get_stream_of_blobhash ( self , blob_hash ) : <EOL> d = self . db_conn . runQuery ( "<STR_LIT>" , <EOL> ( blob_hash , ) ) <EOL> d . addCallback ( lambda r : r [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] if len ( r ) else None ) <EOL> return d <EOL> @ rerun_if_locked <EOL> def _save_sd_blob_hash_to_stream ( self , stream_hash , sd_blob_hash ) : <EOL> log . info ( "<STR_LIT>" , str ( sd_blob_hash ) , str ( stream_hash ) ) <EOL> d = self . db_conn . runQuery ( "<STR_LIT>" , <EOL> ( sd_blob_hash , stream_hash ) ) <EOL> def ignore_duplicate ( err ) : <EOL> err . trap ( sqlite3 . IntegrityError ) <EOL> log . info ( "<STR_LIT>" ) <EOL> d . addErrback ( ignore_duplicate ) <EOL> return d <EOL> @ rerun_if_locked <EOL> def _get_sd_blob_hashes_for_stream ( self , stream_hash ) : <EOL> log . info ( "<STR_LIT>" , str ( stream_hash ) ) <EOL> d = self . db_conn . runQuery ( "<STR_LIT>" , <EOL> ( stream_hash , ) ) <EOL> d . addCallback ( lambda results : [ r [ <NUM_LIT:0> ] for r in results ] ) <EOL> return d <EOL> class TempLBRYFileMetadataManager ( object ) : <EOL> def __init__ ( self ) : <EOL> self . streams = { } <EOL> self . stream_blobs = { } <EOL> self . sd_files = { } <EOL> def setup ( self ) : <EOL> return defer . succeed ( True ) <EOL> def stop ( self ) : <EOL> return defer . succeed ( True ) <EOL> def get_all_streams ( self ) : <EOL> return defer . succeed ( self . streams . keys ( ) ) <EOL> def save_stream ( self , stream_hash , file_name , key , suggested_file_name , blobs ) : <EOL> self . streams [ stream_hash ] = { '<STR_LIT>' : suggested_file_name , <EOL> '<STR_LIT>' : file_name , <EOL> '<STR_LIT:key>' : key } <EOL> d = self . add_blobs_to_stream ( stream_hash , blobs ) <EOL> d . addCallback ( lambda _ : stream_hash ) <EOL> return d <EOL> def get_stream_info ( self , stream_hash ) : <EOL> if stream_hash in self . streams : <EOL> stream_info = self . streams [ stream_hash ] <EOL> return defer . succeed ( [ stream_info [ '<STR_LIT:key>' ] , stream_info [ '<STR_LIT>' ] , <EOL> stream_info [ '<STR_LIT>' ] ] ) <EOL> return defer . succeed ( None ) <EOL> def delete_stream ( self , stream_hash ) : <EOL> if stream_hash in self . streams : <EOL> del self . streams [ stream_hash ] <EOL> for ( s_h , b_h ) in self . stream_blobs . keys ( ) : <EOL> if s_h == stream_hash : <EOL> del self . stream_blobs [ ( s_h , b_h ) ] <EOL> return defer . succeed ( True ) <EOL> def add_blobs_to_stream ( self , stream_hash , blobs ) : <EOL> assert stream_hash in self . streams , "<STR_LIT>" <EOL> for blob in blobs : <EOL> info = { } <EOL> info [ '<STR_LIT>' ] = blob . blob_num <EOL> info [ '<STR_LIT>' ] = blob . length <EOL> info [ '<STR_LIT>' ] = blob . iv <EOL> self . stream_blobs [ ( stream_hash , blob . blob_hash ) ] = info <EOL> return defer . succeed ( True ) <EOL> def get_blobs_for_stream ( self , stream_hash , start_blob = None , end_blob = None , count = None , reverse = False ) : <EOL> if start_blob is not None : <EOL> start_num = self . _get_blob_num_by_hash ( stream_hash , start_blob ) <EOL> else : <EOL> start_num = None <EOL> if end_blob is not None : <EOL> end_num = self . _get_blob_num_by_hash ( stream_hash , end_blob ) <EOL> else : <EOL> end_num = None <EOL> return self . _get_further_blob_infos ( stream_hash , start_num , end_num , count , reverse ) <EOL> def get_stream_of_blob ( self , blob_hash ) : <EOL> for ( s_h , b_h ) in self . stream_blobs . iterkeys ( ) : <EOL> if b_h == blob_hash : <EOL> return defer . succeed ( s_h ) <EOL> return defer . succeed ( None ) <EOL> def _get_further_blob_infos ( self , stream_hash , start_num , end_num , count = None , reverse = False ) : <EOL> blob_infos = [ ] <EOL> for ( s_h , b_h ) , info in self . stream_blobs . iteritems ( ) : <EOL> if stream_hash == s_h : <EOL> position = info [ '<STR_LIT>' ] <EOL> length = info [ '<STR_LIT>' ] <EOL> iv = info [ '<STR_LIT>' ] <EOL> if ( start_num is None ) or ( position > start_num ) : <EOL> if ( end_num is None ) or ( position < end_num ) : <EOL> blob_infos . append ( ( b_h , position , iv , length ) ) <EOL> blob_infos . sort ( key = lambda i : i [ <NUM_LIT:1> ] , reverse = reverse ) <EOL> if count is not None : <EOL> blob_infos = blob_infos [ : count ] <EOL> return defer . succeed ( blob_infos ) <EOL> def _get_blob_num_by_hash ( self , stream_hash , blob_hash ) : <EOL> if ( stream_hash , blob_hash ) in self . stream_blobs : <EOL> return defer . succeed ( self . stream_blobs [ ( stream_hash , blob_hash ) ] [ '<STR_LIT>' ] ) <EOL> def save_sd_blob_hash_to_stream ( self , stream_hash , sd_blob_hash ) : <EOL> self . sd_files [ sd_blob_hash ] = stream_hash <EOL> return defer . succeed ( True ) <EOL> def get_sd_blob_hashes_for_stream ( self , stream_hash ) : <EOL> return defer . succeed ( [ sd_hash for sd_hash , s_h in self . sd_files . iteritems ( ) if stream_hash == s_h ] ) </s>
