<s> import collections <EOL> import json <EOL> import logging <EOL> import re <EOL> import urllib <EOL> import urllib2 <EOL> import base . cache <EOL> import base . paths <EOL> import base . url_fetcher <EOL> FEED_STREAM_ID_PREFIX = '<STR_LIT>' <EOL> EXPLORE_STREAM_ID = '<STR_LIT>' <EOL> _ITEM_ID_ATOM_FORM_PREFIX = '<STR_LIT>' <EOL> not_found_items_ids_to_ignore = set ( ) <EOL> class Api ( object ) : <EOL> def __init__ ( self , <EOL> authenticated_url_fetcher , http_retry_count = None , cache_directory = None ) : <EOL> self . _direct_url_fetcher = base . url_fetcher . DirectUrlFetcher ( ) <EOL> self . _authenticated_url_fetcher = authenticated_url_fetcher <EOL> if http_retry_count > <NUM_LIT:1> : <EOL> self . _direct_url_fetcher = base . url_fetcher . RetryingUrlFetcher ( <EOL> http_retry_count , self . _direct_url_fetcher ) <EOL> self . _authenticated_url_fetcher = base . url_fetcher . RetryingUrlFetcher ( <EOL> http_retry_count , self . _authenticated_url_fetcher ) <EOL> self . _cache = base . cache . DirectoryCache ( cache_directory ) if cache_directory else None <EOL> def fetch_user_info ( self ) : <EOL> user_info_json = self . _fetch_json ( '<STR_LIT>' ) <EOL> return UserInfo ( <EOL> user_id = user_info_json [ '<STR_LIT>' ] , <EOL> email = user_info_json [ '<STR_LIT>' ] , <EOL> profile_id = user_info_json [ '<STR_LIT>' ] , <EOL> user_name = user_info_json [ '<STR_LIT>' ] , <EOL> public_user_name = user_info_json . get ( '<STR_LIT>' ) , <EOL> is_blogger_user = user_info_json [ '<STR_LIT>' ] , <EOL> signup_time_sec = user_info_json [ '<STR_LIT>' ] , <EOL> is_multi_login_enabled = user_info_json [ '<STR_LIT>' ] ) <EOL> def fetch_tags ( self ) : <EOL> tags_json = self . _fetch_json ( '<STR_LIT>' ) <EOL> result = [ ] <EOL> for tag_json in tags_json [ '<STR_LIT>' ] : <EOL> result . append ( Tag ( <EOL> stream_id = tag_json [ '<STR_LIT:id>' ] , <EOL> sort_id = tag_json [ '<STR_LIT>' ] <EOL> ) ) <EOL> return result <EOL> def fetch_subscriptions ( self ) : <EOL> subscriptions_json = self . _fetch_json ( '<STR_LIT>' ) <EOL> result = [ ] <EOL> for subscription_json in subscriptions_json [ '<STR_LIT>' ] : <EOL> insert_stream_ids = [ ] <EOL> for category_json in subscription_json . get ( '<STR_LIT>' , [ ] ) : <EOL> insert_stream_ids . append ( category_json [ '<STR_LIT:id>' ] ) <EOL> result . append ( Subscription ( <EOL> stream_id = subscription_json [ '<STR_LIT:id>' ] , <EOL> sort_id = subscription_json [ '<STR_LIT>' ] , <EOL> title = subscription_json . get ( '<STR_LIT:title>' ) , <EOL> first_item_usec = int ( subscription_json . get ( '<STR_LIT>' , <NUM_LIT:0> ) ) * <NUM_LIT:1000> , <EOL> html_url = subscription_json . get ( '<STR_LIT>' ) , <EOL> insert_stream_ids = insert_stream_ids , <EOL> ) ) <EOL> return result <EOL> def fetch_friends ( self ) : <EOL> friends_json = self . _fetch_json ( '<STR_LIT>' , { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> result = [ ] <EOL> for friend_json in friends_json [ '<STR_LIT>' ] : <EOL> flags = friend_json . get ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> types = friend_json . get ( '<STR_LIT>' , [ ] ) <EOL> websites = [ <EOL> Website ( w [ '<STR_LIT:title>' ] , w [ '<STR_LIT:url>' ] ) for w in friend_json . get ( '<STR_LIT>' , [ ] ) ] <EOL> result . append ( Friend ( <EOL> stream_id = friend_json . get ( '<STR_LIT>' ) , <EOL> user_ids = friend_json . get ( '<STR_LIT>' , [ ] ) , <EOL> profile_ids = friend_json . get ( '<STR_LIT>' , [ ] ) , <EOL> contact_id = friend_json [ '<STR_LIT>' ] , <EOL> group_ids = friend_json . get ( '<STR_LIT>' , [ ] ) , <EOL> display_name = friend_json [ '<STR_LIT>' ] , <EOL> given_name = friend_json [ '<STR_LIT>' ] , <EOL> occupation = friend_json . get ( '<STR_LIT>' ) , <EOL> websites = websites , <EOL> location = friend_json . get ( '<STR_LIT:location>' ) , <EOL> photo_url = friend_json . get ( '<STR_LIT>' ) , <EOL> email_addresses = friend_json . get ( '<STR_LIT>' , [ ] ) , <EOL> is_current_user = flags & <NUM_LIT:1> << <NUM_LIT:0> != <NUM_LIT:0> , <EOL> is_hidden = flags & <NUM_LIT:1> << <NUM_LIT:1> != <NUM_LIT:0> , <EOL> is_new = flags & <NUM_LIT:1> << <NUM_LIT:2> != <NUM_LIT:0> , <EOL> uses_reader = flags & <NUM_LIT:1> << <NUM_LIT:3> != <NUM_LIT:0> , <EOL> is_blocked = flags & <NUM_LIT:1> << <NUM_LIT:4> != <NUM_LIT:0> , <EOL> has_profile = flags & <NUM_LIT:1> << <NUM_LIT:5> != <NUM_LIT:0> , <EOL> is_ignored = flags & <NUM_LIT:1> << <NUM_LIT:6> != <NUM_LIT:0> , <EOL> is_new_follower = flags & <NUM_LIT:1> << <NUM_LIT:7> != <NUM_LIT:0> , <EOL> is_anonymous = flags & <NUM_LIT:1> << <NUM_LIT:8> != <NUM_LIT:0> , <EOL> has_shared_items = flags & <NUM_LIT:1> << <NUM_LIT:9> != <NUM_LIT:0> , <EOL> is_follower = <NUM_LIT:0> in types , <EOL> is_following = <NUM_LIT:1> in types , <EOL> is_contact = <NUM_LIT:3> in types , <EOL> is_pending_following = <NUM_LIT:4> in types , <EOL> is_pending_follower = <NUM_LIT:5> in types , <EOL> is_allowed_following = <NUM_LIT:6> in types , <EOL> is_allowed_commenting = <NUM_LIT:7> in types , <EOL> ) ) <EOL> return result <EOL> def fetch_encoded_sharers ( self ) : <EOL> friends_json = self . _fetch_json ( '<STR_LIT>' , { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> return friends_json . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> def fetch_sharing_groups ( self ) : <EOL> sharing_groups_json = self . _fetch_json ( '<STR_LIT>' ) <EOL> result = [ ] <EOL> for sharing_group_json in sharing_groups_json [ '<STR_LIT>' ] : <EOL> result . append ( SharingGroup ( <EOL> group_id = sharing_group_json [ '<STR_LIT>' ] , <EOL> is_read_only = sharing_group_json [ '<STR_LIT>' ] , <EOL> name = sharing_group_json [ '<STR_LIT:name>' ] , <EOL> is_sharing = sharing_group_json [ '<STR_LIT>' ] , <EOL> ) ) <EOL> return result <EOL> def fetch_sharing_acl ( self ) : <EOL> sharing_acl_json = self . _fetch_json ( '<STR_LIT>' ) <EOL> return SharingAcl ( <EOL> type = sharing_acl_json [ '<STR_LIT:type>' ] , <EOL> member_user_ids = sharing_acl_json [ '<STR_LIT>' ] , <EOL> is_editing_disabled = sharing_acl_json [ '<STR_LIT>' ] ) <EOL> def fetch_bundles ( self ) : <EOL> bundles_json = self . _fetch_json ( '<STR_LIT>' ) <EOL> result = [ ] <EOL> for bundle_json in bundles_json [ '<STR_LIT>' ] : <EOL> feeds = [ ] <EOL> for feed_json in bundle_json [ '<STR_LIT>' ] : <EOL> feeds . append ( BundleFeed ( <EOL> stream_id = feed_json [ '<STR_LIT:id>' ] , title = feed_json [ '<STR_LIT:title>' ] ) ) <EOL> result . append ( Bundle ( <EOL> bundle_id = bundle_json [ '<STR_LIT:id>' ] , <EOL> title = bundle_json [ '<STR_LIT:title>' ] , <EOL> description = bundle_json . get ( '<STR_LIT:description>' ) , <EOL> subscriber_count = bundle_json [ '<STR_LIT>' ] , <EOL> feeds = feeds ) ) <EOL> return result <EOL> def fetch_recommendations ( self , count = <NUM_LIT:20> ) : <EOL> recommendations_json = self . _fetch_json ( <EOL> '<STR_LIT>' , { '<STR_LIT:n>' : count } ) <EOL> result = [ ] <EOL> for recommendation_json in recommendations_json [ '<STR_LIT>' ] : <EOL> result . append ( Recommendation ( <EOL> stream_id = recommendation_json [ '<STR_LIT>' ] , <EOL> title = recommendation_json [ '<STR_LIT:title>' ] ) ) <EOL> return result <EOL> def fetch_preferences ( self ) : <EOL> prefs_json = self . _fetch_json ( '<STR_LIT>' ) <EOL> result = { } <EOL> for pref_json in prefs_json [ '<STR_LIT>' ] : <EOL> result [ pref_json [ '<STR_LIT:id>' ] ] = pref_json [ '<STR_LIT:value>' ] <EOL> return result <EOL> def fetch_stream_preferences ( self ) : <EOL> prefs_json = self . _fetch_json ( '<STR_LIT>' ) <EOL> result = { } <EOL> for stream_id , stream_prefs_json in prefs_json [ '<STR_LIT>' ] . iteritems ( ) : <EOL> stream_prefs = { } <EOL> for pref_json in stream_prefs_json : <EOL> stream_prefs [ pref_json [ '<STR_LIT:id>' ] ] = pref_json [ '<STR_LIT:value>' ] <EOL> result [ stream_id ] = stream_prefs <EOL> return result <EOL> def fetch_item_refs ( self , stream_id , count = <NUM_LIT:10> , continuation_token = None ) : <EOL> query_params = { '<STR_LIT:s>' : stream_id , '<STR_LIT:n>' : count } <EOL> if continuation_token : <EOL> query_params [ '<STR_LIT:c>' ] = continuation_token <EOL> item_refs_json = self . _fetch_json ( <EOL> '<STR_LIT>' , <EOL> query_params , <EOL> authenticated = not stream_id . startswith ( FEED_STREAM_ID_PREFIX ) ) <EOL> result = [ ] <EOL> for item_ref_json in item_refs_json [ '<STR_LIT>' ] : <EOL> result . append ( ItemRef ( <EOL> item_id = item_id_from_decimal_form ( item_ref_json [ '<STR_LIT:id>' ] ) , <EOL> timestamp_usec = int ( item_ref_json [ '<STR_LIT>' ] ) <EOL> ) ) <EOL> return result , item_refs_json . get ( '<STR_LIT>' ) <EOL> def fetch_comments ( <EOL> self , stream_id , encoded_sharers , count = <NUM_LIT:10> , continuation_token = None ) : <EOL> query_params = { <EOL> '<STR_LIT>' : '<STR_LIT:true>' , <EOL> '<STR_LIT>' : encoded_sharers , <EOL> '<STR_LIT:n>' : count , <EOL> } <EOL> if continuation_token : <EOL> query_params [ '<STR_LIT:c>' ] = continuation_token <EOL> stream_contents_json = self . _fetch_json ( <EOL> '<STR_LIT>' % urllib . quote ( stream_id ) , <EOL> query_params ) <EOL> result = { } <EOL> for item_json in stream_contents_json [ '<STR_LIT>' ] : <EOL> comments_json = item_json . get ( '<STR_LIT>' , [ ] ) <EOL> if not comments_json : <EOL> continue <EOL> item_id = item_id_from_atom_form ( item_json [ '<STR_LIT:id>' ] ) <EOL> comments = [ ] <EOL> for comment_json in comments_json : <EOL> comments . append ( Comment ( <EOL> comment_id = comment_json [ '<STR_LIT:id>' ] , <EOL> plain_content = comment_json [ '<STR_LIT>' ] , <EOL> html_content = comment_json [ '<STR_LIT>' ] , <EOL> author_name = comment_json . get ( '<STR_LIT>' ) , <EOL> author_user_id = comment_json [ '<STR_LIT>' ] , <EOL> author_profile_id = comment_json [ '<STR_LIT>' ] , <EOL> venue_stream_id = comment_json [ '<STR_LIT>' ] , <EOL> created_time_usec = comment_json [ '<STR_LIT>' ] * <NUM_LIT> , <EOL> modified_time_usec = comment_json [ '<STR_LIT>' ] * <NUM_LIT> , <EOL> is_spam = comment_json [ '<STR_LIT>' ] , <EOL> ) ) <EOL> result [ item_id ] = comments <EOL> return result , stream_contents_json . get ( '<STR_LIT>' ) <EOL> def fetch_item_bodies ( <EOL> self , item_ids , format = '<STR_LIT>' , media_rss = False , authenticated = True ) : <EOL> query_params = { <EOL> '<STR_LIT>' : format , <EOL> '<STR_LIT>' : '<STR_LIT:false>' , <EOL> '<STR_LIT>' : '<STR_LIT:true>' , <EOL> } <EOL> if media_rss : <EOL> query_params [ '<STR_LIT>' ] = '<STR_LIT:true>' <EOL> post_params = { '<STR_LIT:i>' : [ i . decimal_form for i in item_ids ] } <EOL> result_text = self . _fetch ( <EOL> '<STR_LIT>' , <EOL> query_params , <EOL> post_params , <EOL> authenticated = authenticated ) <EOL> result = { } <EOL> if format . startswith ( '<STR_LIT>' ) : <EOL> feed = base . atom . parse ( result_text ) <EOL> for entry in feed . entries : <EOL> result [ entry . item_id ] = entry <EOL> else : <EOL> item_bodies_json = json . loads ( result_text ) <EOL> for item_body_json in item_bodies_json [ '<STR_LIT>' ] : <EOL> item_id = item_id_from_atom_form ( item_body_json [ '<STR_LIT:id>' ] ) <EOL> result [ item_id ] = item_body_json <EOL> for item_id in item_ids : <EOL> if item_id not in result and item_id not in not_found_items_ids_to_ignore : <EOL> logging . warning ( <EOL> '<STR_LIT>' , <EOL> item_id . atom_form , item_id . decimal_form ) <EOL> return result <EOL> def _fetch_json ( <EOL> self , <EOL> api_path , <EOL> query_params = { } , <EOL> post_params = { } , <EOL> authenticated = True ) : <EOL> query_params = dict ( query_params ) <EOL> query_params [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> response_text = self . _fetch ( <EOL> api_path , query_params , post_params , authenticated ) <EOL> return json . loads ( response_text ) <EOL> def _fetch ( self , <EOL> api_path , <EOL> query_params = { } , <EOL> post_params = { } , <EOL> authenticated = True ) : <EOL> url = '<STR_LIT>' % api_path <EOL> if self . _cache : <EOL> cache_key = base . paths . url_to_file_name ( url , query_params , post_params ) <EOL> cache_value = self . _cache . get ( cache_key ) <EOL> if cache_value : <EOL> return cache_value <EOL> def urlencode ( params ) : <EOL> def encode ( s ) : <EOL> return isinstance ( s , unicode ) and s . encode ( '<STR_LIT:utf-8>' ) or s <EOL> encoded_params = { } <EOL> for key , value in params . items ( ) : <EOL> if isinstance ( value , list ) : <EOL> value = [ encode ( v ) for v in value ] <EOL> else : <EOL> value = encode ( value ) <EOL> encoded_params [ encode ( key ) ] = value <EOL> return urllib . urlencode ( encoded_params , doseq = True ) <EOL> request_url = '<STR_LIT>' % ( url , urlencode ( query_params ) ) <EOL> url_fetcher = self . _authenticated_url_fetcher if authenticated else self . _direct_url_fetcher <EOL> response_text = url_fetcher . fetch ( <EOL> request_url , <EOL> post_data = urlencode ( post_params ) if post_params else None ) <EOL> if self . _cache : <EOL> self . _cache . set ( cache_key , response_text ) <EOL> return response_text <EOL> class Tag ( collections . namedtuple ( '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT>' ] ) ) : <EOL> def to_json ( self ) : <EOL> return self . _asdict ( ) <EOL> @ staticmethod <EOL> def from_json ( tag_json ) : <EOL> return Tag ( ** tag_json ) <EOL> class Subscription ( collections . namedtuple ( <EOL> '<STR_LIT>' , <EOL> [ '<STR_LIT>' , '<STR_LIT:title>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) ) : <EOL> def to_json ( self ) : <EOL> return self . _asdict ( ) <EOL> @ staticmethod <EOL> def from_json ( subscription_json ) : <EOL> return Subscription ( ** subscription_json ) <EOL> class Friend ( collections . namedtuple ( <EOL> '<STR_LIT>' , <EOL> [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:location>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ] ) ) : <EOL> def to_json ( self ) : <EOL> result = self . _asdict ( ) <EOL> result [ '<STR_LIT>' ] = [ w . to_json ( ) for w in self . websites ] <EOL> return result <EOL> @ staticmethod <EOL> def from_json ( friend_json ) : <EOL> return Friend ( ** friend_json ) <EOL> class Website ( collections . namedtuple ( '<STR_LIT>' , [ '<STR_LIT:title>' , '<STR_LIT:url>' ] ) ) : <EOL> def to_json ( self ) : <EOL> return self . _asdict ( ) <EOL> class SharingGroup ( collections . namedtuple ( '<STR_LIT>' , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:name>' , '<STR_LIT>' ] ) ) : <EOL> def to_json ( self ) : <EOL> return self . _asdict ( ) <EOL> class SharingAcl ( collections . namedtuple ( '<STR_LIT>' , <EOL> [ '<STR_LIT:type>' , '<STR_LIT>' , '<STR_LIT>' ] ) ) : <EOL> def to_json ( self ) : <EOL> return self . _asdict ( ) <EOL> class Bundle ( collections . namedtuple ( '<STR_LIT>' , <EOL> [ '<STR_LIT>' , '<STR_LIT:title>' , '<STR_LIT:description>' , '<STR_LIT>' , '<STR_LIT>' ] ) ) : <EOL> def to_json ( self ) : <EOL> result = self . _asdict ( ) <EOL> result [ '<STR_LIT>' ] = [ f . to_json ( ) for f in self . feeds ] <EOL> return result <EOL> class BundleFeed ( collections . namedtuple ( '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT:title>' ] ) ) : <EOL> def to_json ( self ) : <EOL> return self . _asdict ( ) <EOL> class Recommendation ( collections . namedtuple ( '<STR_LIT>' , <EOL> [ '<STR_LIT>' , '<STR_LIT:title>' ] ) ) : <EOL> def to_json ( self ) : <EOL> return self . _asdict ( ) <EOL> @ staticmethod <EOL> def from_json ( recommendation_json ) : <EOL> return Recommendation ( ** recommendation_json ) <EOL> class Comment ( collections . namedtuple ( <EOL> '<STR_LIT>' , <EOL> [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ] ) ) : <EOL> def to_json ( self ) : <EOL> return self . _asdict ( ) <EOL> @ staticmethod <EOL> def from_json ( comment_json ) : <EOL> return Comment ( ** comment_json ) <EOL> class UserInfo ( collections . namedtuple ( <EOL> '<STR_LIT>' , <EOL> [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT:email>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> ] ) ) : <EOL> def to_json ( self ) : <EOL> return self . _asdict ( ) <EOL> @ staticmethod <EOL> def from_json ( user_info_json ) : <EOL> return UserInfo ( ** user_info_json ) <EOL> class ItemRef ( collections . namedtuple ( '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT>' ] ) ) : <EOL> def to_json ( self ) : <EOL> return { <EOL> '<STR_LIT>' : self . item_id . to_json ( ) , <EOL> '<STR_LIT>' : self . timestamp_usec , <EOL> } <EOL> class Stream ( collections . namedtuple ( '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT>' ] ) ) : <EOL> def to_json ( self ) : <EOL> return { <EOL> '<STR_LIT>' : self . stream_id , <EOL> '<STR_LIT>' : { <EOL> item_ref . item_id . to_json ( ) : item_ref . timestamp_usec <EOL> for item_ref in self . item_refs <EOL> } , <EOL> } <EOL> @ staticmethod <EOL> def from_json ( stream_json ) : <EOL> item_refs = [ <EOL> ItemRef ( <EOL> item_id = ItemId . from_json ( item_id_json ) , <EOL> timestamp_usec = timestamp_usec <EOL> ) for item_id_json , timestamp_usec <EOL> in stream_json [ '<STR_LIT>' ] . iteritems ( ) <EOL> ] <EOL> item_refs = sorted ( item_refs , key = lambda i : i . timestamp_usec , reverse = True ) <EOL> return Stream ( stream_id = stream_json [ '<STR_LIT>' ] , item_refs = item_refs ) <EOL> class ItemId ( collections . namedtuple ( '<STR_LIT>' , [ '<STR_LIT>' ] ) ) : <EOL> def to_json ( self ) : <EOL> return self . compact_form ( ) <EOL> def compact_form ( self ) : <EOL> compact_form = hex ( self . int_form ) [ <NUM_LIT:2> : ] <EOL> if compact_form . endswith ( '<STR_LIT:L>' ) : <EOL> compact_form = compact_form [ : - <NUM_LIT:1> ] <EOL> compact_form = ( <NUM_LIT:16> - len ( compact_form ) ) * '<STR_LIT:0>' + compact_form <EOL> return compact_form <EOL> @ property <EOL> def decimal_form ( self ) : <EOL> if self . int_form > <NUM_LIT:1> << <NUM_LIT> : <EOL> return str ( self . int_form - ( <NUM_LIT:1> << <NUM_LIT:64> ) ) <EOL> else : <EOL> return str ( self . int_form ) <EOL> @ property <EOL> def atom_form ( self ) : <EOL> return _ITEM_ID_ATOM_FORM_PREFIX + self . compact_form ( ) <EOL> @ staticmethod <EOL> def from_json ( item_id_json ) : <EOL> return item_id_from_compact_form ( item_id_json ) <EOL> def item_id_from_decimal_form ( decimal_form ) : <EOL> int_form = int ( decimal_form ) <EOL> if int_form < <NUM_LIT:0> : <EOL> int_form += <NUM_LIT:1> << <NUM_LIT:64> <EOL> return ItemId ( int_form = int_form ) <EOL> def item_id_from_atom_form ( atom_form ) : <EOL> return item_id_from_compact_form ( atom_form [ len ( _ITEM_ID_ATOM_FORM_PREFIX ) : ] ) <EOL> def item_id_from_compact_form ( compact_form ) : <EOL> return ItemId ( int_form = int ( compact_form , <NUM_LIT:16> ) ) <EOL> def item_id_from_any_form ( form ) : <EOL> if form . startswith ( _ITEM_ID_ATOM_FORM_PREFIX ) : <EOL> return item_id_from_atom_form ( form ) <EOL> if form . startswith ( '<STR_LIT>' ) : <EOL> return item_id_from_compact_form ( form [ <NUM_LIT:2> : ] ) <EOL> if re . match ( '<STR_LIT>' , form , re . I ) : <EOL> return item_id_from_compact_form ( form ) <EOL> if re . match ( '<STR_LIT>' , form ) : <EOL> return item_id_from_decimal_form ( form ) <EOL> return None <EOL> _TEST_DATA = [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ] <EOL> def _test_ids ( ) : <EOL> for atom_form , decimal_form in _TEST_DATA : <EOL> item_id = item_id_from_decimal_form ( decimal_form ) <EOL> assert item_id . atom_form == atom_form , '<STR_LIT>' % ( item_id . atom_form , atom_form ) <EOL> item_id = item_id_from_atom_form ( atom_form ) <EOL> assert item_id . decimal_form == decimal_form , '<STR_LIT>' % ( item_id . decimal_form , decimal_form ) </s>
