<s> """<STR_LIT>""" <EOL> from . compat import json <EOL> from . helpers import is_collection <EOL> from . models import Plan , RateLimit <EOL> from . models . app import App <EOL> from . models . addon import Addon <EOL> from . models . dyno import Dyno <EOL> from . models . account import Account <EOL> from . models . key import Key <EOL> from . models . invoice import Invoice <EOL> from . models . configvars import ConfigVars <EOL> from . models . logsession import LogSession <EOL> from . models . oauth import OAuthClient , OAuthAuthorization , OAuthToken <EOL> from . rendezvous import Rendezvous <EOL> from . structures import KeyedListResource , SSHKeyListResource <EOL> from . models . account . feature import AccountFeature <EOL> from requests . exceptions import HTTPError <EOL> from pprint import pprint <EOL> import requests <EOL> import sys <EOL> if sys . version_info > ( <NUM_LIT:3> , <NUM_LIT:0> ) : <EOL> from urllib . parse import quote <EOL> else : <EOL> from urllib import quote <EOL> HEROKU_URL = '<STR_LIT>' <EOL> class RateLimitExceeded ( Exception ) : <EOL> pass <EOL> class MaxRangeExceeded ( Exception ) : <EOL> pass <EOL> class HerokuCore ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , session = None ) : <EOL> super ( HerokuCore , self ) . __init__ ( ) <EOL> if session is None : <EOL> session = requests . session ( ) <EOL> self . _api_key = None <EOL> self . _api_key_verified = None <EOL> self . _heroku_url = HEROKU_URL <EOL> self . _session = session <EOL> self . _ratelimit_remaining = None <EOL> self . _last_request_id = None <EOL> self . _session . headers . update ( { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:Content-Type>' : '<STR_LIT:application/json>' } ) <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % ( id ( self ) ) <EOL> def authenticate ( self , api_key ) : <EOL> """<STR_LIT>""" <EOL> self . _api_key = api_key <EOL> self . _session . auth = ( '<STR_LIT>' , self . _api_key ) <EOL> return self . _verify_api_key ( ) <EOL> @ property <EOL> def is_authenticated ( self ) : <EOL> if self . _api_key_verified is None : <EOL> return self . _verify_api_key ( ) <EOL> else : <EOL> return self . _api_key_verified <EOL> def _verify_api_key ( self ) : <EOL> r = self . _session . get ( self . _url_for ( '<STR_LIT>' ) ) <EOL> self . _api_key_verified = True if r . ok else False <EOL> return self . _api_key_verified <EOL> def _url_for ( self , * args ) : <EOL> args = map ( str , args ) <EOL> return '<STR_LIT:/>' . join ( [ self . _heroku_url ] + list ( args ) ) <EOL> @ staticmethod <EOL> def _resource_serialize ( o ) : <EOL> """<STR_LIT>""" <EOL> return json . dumps ( o ) <EOL> @ staticmethod <EOL> def _resource_deserialize ( s ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return json . loads ( s ) <EOL> except ValueError : <EOL> raise ResponseError ( '<STR_LIT>' ) <EOL> def _get_headers_for_request ( self , method , url , legacy = False , order_by = None , limit = None , valrange = None , sort = None ) : <EOL> headers = { } <EOL> if legacy is True : <EOL> headers . update ( { '<STR_LIT>' : '<STR_LIT:application/json>' } ) <EOL> else : <EOL> range_str = None <EOL> if order_by or limit or valrange or sort : <EOL> range_str = "<STR_LIT>" <EOL> seperator = '<STR_LIT>' <EOL> if order_by : <EOL> range_str = "<STR_LIT>" . format ( order_by ) <EOL> else : <EOL> range_str = "<STR_LIT>" <EOL> if not sort is None : <EOL> assert ( sort == '<STR_LIT>' or sort == '<STR_LIT>' ) <EOL> range_str += "<STR_LIT>" . format ( sort ) <EOL> seperator = '<STR_LIT:U+002C>' <EOL> else : <EOL> range_str += '<STR_LIT:U+0020>' <EOL> if limit : <EOL> if limit > <NUM_LIT:1000> : <EOL> raise MaxRangeExceeded ( "<STR_LIT>" . format ( limit ) ) <EOL> range_str += "<STR_LIT>" . format ( seperator , limit ) <EOL> range_str += '<STR_LIT:;>' <EOL> if valrange : <EOL> range_str = valrange <EOL> if not range_str == None : <EOL> headers . update ( { '<STR_LIT>' : range_str } ) <EOL> return headers <EOL> def _http_resource ( self , method , resource , params = None , data = None , legacy = False , order_by = None , limit = None , valrange = None , sort = None ) : <EOL> """<STR_LIT>""" <EOL> if not is_collection ( resource ) : <EOL> resource = [ resource ] <EOL> url = self . _url_for ( * resource ) <EOL> headers = self . _get_headers_for_request ( method , url , legacy = legacy , order_by = order_by , limit = limit , valrange = valrange , sort = sort ) <EOL> r = self . _session . request ( method , url , params = params , data = data , headers = headers ) <EOL> if '<STR_LIT>' in r . headers : <EOL> self . _ratelimit_remaining = r . headers [ '<STR_LIT>' ] <EOL> if '<STR_LIT>' in r . headers : <EOL> self . _last_request_id = r . headers [ '<STR_LIT>' ] <EOL> if r . status_code == <NUM_LIT> : <EOL> http_error = HTTPError ( '<STR_LIT>' % <EOL> ( self . _last_request_id , r . status_code , r . content . decode ( "<STR_LIT:utf-8>" ) ) ) <EOL> http_error . response = r <EOL> raise http_error <EOL> if r . status_code == <NUM_LIT> : <EOL> raise RateLimitExceeded ( "<STR_LIT>" . format ( r . content . decode ( "<STR_LIT:utf-8>" ) ) ) <EOL> if ( not str ( r . status_code ) . startswith ( '<STR_LIT:2>' ) ) and ( not r . status_code in [ <NUM_LIT> ] ) : <EOL> pass <EOL> r . raise_for_status ( ) <EOL> return r <EOL> def _get_resource ( self , resource , obj , params = None , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> r = self . _http_resource ( '<STR_LIT:GET>' , resource , params = params ) <EOL> return self . _process_item ( self . _resource_deserialize ( r . content . decode ( "<STR_LIT:utf-8>" ) ) , obj , ** kwargs ) <EOL> def _process_item ( self , item , obj , ** kwargs ) : <EOL> return obj . new_from_dict ( item , h = self , ** kwargs ) <EOL> def _get_resources ( self , resource , obj , params = None , map = None , legacy = None , order_by = None , limit = None , valrange = None , sort = None , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if not order_by : <EOL> order_by = obj . order_by <EOL> return self . _process_items ( self . _get_data ( resource , params = params , legacy = legacy , order_by = order_by , limit = limit , valrange = valrange , sort = sort ) , obj , map = map , ** kwargs ) <EOL> def _get_data ( self , resource , params = None , legacy = None , order_by = None , limit = None , valrange = None , sort = None ) : <EOL> r = self . _http_resource ( '<STR_LIT:GET>' , resource , params = params , legacy = legacy , order_by = order_by , limit = limit , valrange = valrange , sort = sort ) <EOL> items = self . _resource_deserialize ( r . content . decode ( "<STR_LIT:utf-8>" ) ) <EOL> if r . status_code == <NUM_LIT> and '<STR_LIT>' in r . headers and not limit : <EOL> valrange = r . headers [ '<STR_LIT>' ] <EOL> print ( "<STR_LIT>" . format ( valrange ) ) <EOL> new_items = self . _get_data ( resource , params = params , legacy = legacy , order_by = order_by , limit = limit , valrange = valrange , sort = sort ) <EOL> items . extend ( new_items ) <EOL> return items <EOL> def _process_items ( self , d_items , obj , map = None , ** kwargs ) : <EOL> if not isinstance ( d_items , list ) : <EOL> print ( "<STR_LIT>" . format ( obj , type ( d_items ) ) ) <EOL> if isinstance ( d_items , dict ) : <EOL> print ( "<STR_LIT>" ) <EOL> return self . _process_item ( d_items , obj , ** kwargs ) <EOL> items = [ obj . new_from_dict ( item , h = self , ** kwargs ) for item in d_items ] <EOL> if map is None : <EOL> map = KeyedListResource <EOL> list_resource = map ( items = items ) <EOL> list_resource . _h = self <EOL> list_resource . _obj = obj <EOL> list_resource . _kwargs = kwargs <EOL> return list_resource <EOL> class Heroku ( HerokuCore ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , session = None ) : <EOL> super ( Heroku , self ) . __init__ ( session = session ) <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % ( id ( self ) ) <EOL> def account ( self ) : <EOL> return self . _get_resource ( ( '<STR_LIT>' ) , Account ) <EOL> def addons ( self , app_id_or_name , ** kwargs ) : <EOL> return self . _get_resources ( resource = ( '<STR_LIT>' , app_id_or_name , '<STR_LIT>' ) , obj = Addon , ** kwargs ) <EOL> def addon_services ( self , id_or_name = None , ** kwargs ) : <EOL> if id_or_name is not None : <EOL> return self . _get_resource ( ( '<STR_LIT>' . format ( quote ( id_or_name ) ) ) , Plan ) <EOL> else : <EOL> return self . _get_resources ( ( '<STR_LIT>' ) , Plan , ** kwargs ) <EOL> def apps ( self , ** kwargs ) : <EOL> return self . _get_resources ( ( '<STR_LIT>' ) , App , ** kwargs ) <EOL> def app ( self , id_or_name ) : <EOL> return self . _get_resource ( ( '<STR_LIT>' . format ( id_or_name ) ) , App ) <EOL> def create_app ( self , name = None , stack_id_or_name = '<STR_LIT>' , region_id_or_name = None ) : <EOL> """<STR_LIT>""" <EOL> payload = { } <EOL> if name : <EOL> payload [ '<STR_LIT:name>' ] = name <EOL> if stack_id_or_name : <EOL> payload [ '<STR_LIT>' ] = stack_id_or_name <EOL> if region_id_or_name : <EOL> payload [ '<STR_LIT>' ] = region_id_or_name <EOL> try : <EOL> r = self . _http_resource ( <EOL> method = '<STR_LIT:POST>' , <EOL> resource = ( '<STR_LIT>' , ) , <EOL> data = self . _resource_serialize ( payload ) <EOL> ) <EOL> r . raise_for_status ( ) <EOL> item = self . _resource_deserialize ( r . content . decode ( "<STR_LIT:utf-8>" ) ) <EOL> app = App . new_from_dict ( item , h = self ) <EOL> except HTTPError as e : <EOL> if "<STR_LIT>" in str ( e ) : <EOL> print ( "<STR_LIT>" . format ( e ) ) <EOL> app = self . app ( name ) <EOL> pass <EOL> else : <EOL> raise e <EOL> return app <EOL> def keys ( self , ** kwargs ) : <EOL> return self . _get_resources ( ( '<STR_LIT>' ) , Key , map = SSHKeyListResource , ** kwargs ) <EOL> def invoices ( self , ** kwargs ) : <EOL> return self . _get_resources ( ( '<STR_LIT>' ) , Invoice ) <EOL> def labs ( self , ** kwargs ) : <EOL> return self . features ( ** kwargs ) <EOL> def features ( self , ** kwargs ) : <EOL> return self . _get_resources ( ( '<STR_LIT>' ) , AccountFeature , ** kwargs ) <EOL> def oauthauthorization ( self , oauthauthorization_id ) : <EOL> return self . _get_resource ( ( '<STR_LIT>' , '<STR_LIT>' , oauthauthorization_id ) , OAuthAuthorization ) <EOL> def oauthauthorizations ( self , ** kwargs ) : <EOL> return self . _get_resources ( ( '<STR_LIT>' , '<STR_LIT>' ) , OAuthAuthorization , ** kwargs ) <EOL> def oauthauthorization_create ( self , scope , oauthclient_id = None , description = None ) : <EOL> """<STR_LIT>""" <EOL> payload = { '<STR_LIT>' : scope } <EOL> if oauthclient_id : <EOL> payload . update ( { '<STR_LIT>' : oauthclient_id } ) <EOL> if description : <EOL> payload . update ( { '<STR_LIT:description>' : description } ) <EOL> r = self . _http_resource ( <EOL> method = '<STR_LIT:POST>' , <EOL> resource = ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> data = self . _h . _resource_serialize ( payload ) <EOL> ) <EOL> r . raise_for_status ( ) <EOL> item = self . _resource_deserialize ( r . content . decode ( "<STR_LIT:utf-8>" ) ) <EOL> return OAuthClient . new_from_dict ( item , h = self ) <EOL> def oauthauthorization_delete ( self , oauthauthorization_id ) : <EOL> """<STR_LIT>""" <EOL> r = self . _http_resource ( <EOL> method = '<STR_LIT>' , <EOL> resource = ( '<STR_LIT>' , '<STR_LIT>' , oauthauthorization_id ) <EOL> ) <EOL> r . raise_for_status ( ) <EOL> return r . ok <EOL> def oauthclient ( self , oauthclient_id ) : <EOL> return self . _get_resource ( ( '<STR_LIT>' , '<STR_LIT>' , oauthclient_id ) , OAuthClient ) <EOL> def oauthclients ( self , ** kwargs ) : <EOL> return self . _get_resources ( ( '<STR_LIT>' , '<STR_LIT>' ) , OAuthClient , ** kwargs ) <EOL> def oauthclient_create ( self , name , redirect_uri ) : <EOL> """<STR_LIT>""" <EOL> payload = { '<STR_LIT:name>' : name , '<STR_LIT>' : redirect_uri } <EOL> r = self . _http_resource ( <EOL> method = '<STR_LIT:POST>' , <EOL> resource = ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> data = self . _h . _resource_serialize ( payload ) <EOL> ) <EOL> r . raise_for_status ( ) <EOL> item = self . _resource_deserialize ( r . content . decode ( "<STR_LIT:utf-8>" ) ) <EOL> return OAuthClient . new_from_dict ( item , h = self ) <EOL> def oauthclient_delete ( self , oauthclient_id ) : <EOL> """<STR_LIT>""" <EOL> r = self . _http_resource ( <EOL> method = '<STR_LIT>' , <EOL> resource = ( '<STR_LIT>' , '<STR_LIT>' , oauthclient_id ) <EOL> ) <EOL> r . raise_for_status ( ) <EOL> return r . ok <EOL> def oauthtoken_create ( self , client_secret = None , grant_code = None , grant_type = None , refresh_token = None ) : <EOL> """<STR_LIT>""" <EOL> payload = { } <EOL> grant = { } <EOL> if client_secret : <EOL> payload . update ( { '<STR_LIT>' : { '<STR_LIT>' : client_secret } } ) <EOL> if grant_code : <EOL> grant . update ( { '<STR_LIT:code>' : grant_code } ) <EOL> if grant_type : <EOL> grant . update ( { '<STR_LIT:type>' : grant_type } ) <EOL> if refresh_token : <EOL> payload . update ( { '<STR_LIT>' : { '<STR_LIT>' : refresh_token } } ) <EOL> if grant : <EOL> payload . update ( { '<STR_LIT>' : grant } ) <EOL> r = self . _http_resource ( <EOL> method = '<STR_LIT:POST>' , <EOL> resource = ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> data = self . _h . _resource_serialize ( payload ) <EOL> ) <EOL> r . raise_for_status ( ) <EOL> item = self . _resource_deserialize ( r . content . decode ( "<STR_LIT:utf-8>" ) ) <EOL> return OAuthToken . new_from_dict ( item , h = self ) <EOL> def run_command_on_app ( self , appname , command , size = <NUM_LIT:1> , attach = True , printout = True , env = None ) : <EOL> """<STR_LIT>""" <EOL> if attach : <EOL> attach = True <EOL> payload = { '<STR_LIT>' : command , '<STR_LIT>' : attach , '<STR_LIT:size>' : size } <EOL> if env : <EOL> payload [ '<STR_LIT>' ] = env <EOL> r = self . _http_resource ( <EOL> method = '<STR_LIT:POST>' , <EOL> resource = ( '<STR_LIT>' , appname , '<STR_LIT>' ) , <EOL> data = self . _resource_serialize ( payload ) <EOL> ) <EOL> r . raise_for_status ( ) <EOL> item = self . _resource_deserialize ( r . content . decode ( "<STR_LIT:utf-8>" ) ) <EOL> dyno = Dyno . new_from_dict ( item , h = self ) <EOL> if attach : <EOL> output = Rendezvous ( dyno . attach_url , printout ) . start ( ) <EOL> return output , dyno <EOL> else : <EOL> return dyno <EOL> @ property <EOL> def rate_limit ( self ) : <EOL> return self . _get_resource ( ( '<STR_LIT>' ) , RateLimit ) <EOL> def ratelimit_remaining ( self ) : <EOL> if self . _ratelimit_remaining is not None : <EOL> return int ( self . _ratelimit_remaining ) <EOL> else : <EOL> self . rate_limit <EOL> return int ( self . _ratelimit_remaining ) <EOL> def stream_app_log ( self , app_id_or_name , dyno = None , lines = <NUM_LIT:100> , source = None , timeout = False ) : <EOL> logger = self . _app_logger ( app_id_or_name , dyno = dyno , lines = lines , source = source , tail = True ) <EOL> return logger . stream ( timeout = timeout ) <EOL> def get_app_log ( self , app_id_or_name , dyno = None , lines = <NUM_LIT:100> , source = None , timeout = False ) : <EOL> logger = self . _app_logger ( app_id_or_name , dyno = dyno , lines = lines , source = source , tail = <NUM_LIT:0> ) <EOL> return logger . get ( timeout = timeout ) <EOL> def update_appconfig ( self , app_id_or_name , config ) : <EOL> payload = self . _resource_serialize ( config ) <EOL> r = self . _http_resource ( <EOL> method = '<STR_LIT>' , <EOL> resource = ( '<STR_LIT>' , app_id_or_name , '<STR_LIT>' ) , <EOL> data = payload <EOL> ) <EOL> r . raise_for_status ( ) <EOL> item = self . _resource_deserialize ( r . content . decode ( "<STR_LIT:utf-8>" ) ) <EOL> return ConfigVars . new_from_dict ( item , h = self ) <EOL> def _app_logger ( self , app_id_or_name , dyno = None , lines = <NUM_LIT:100> , source = None , tail = <NUM_LIT:0> ) : <EOL> payload = { } <EOL> if dyno : <EOL> payload [ '<STR_LIT>' ] = dyno <EOL> if tail : <EOL> payload [ '<STR_LIT>' ] = tail <EOL> if source : <EOL> payload [ '<STR_LIT:source>' ] = source <EOL> if lines : <EOL> payload [ '<STR_LIT>' ] = lines <EOL> r = self . _http_resource ( <EOL> method = '<STR_LIT:POST>' , <EOL> resource = ( '<STR_LIT>' , app_id_or_name , '<STR_LIT>' ) , <EOL> data = self . _resource_serialize ( payload ) <EOL> ) <EOL> r . raise_for_status ( ) <EOL> item = self . _resource_deserialize ( r . content . decode ( "<STR_LIT:utf-8>" ) ) <EOL> return LogSession . new_from_dict ( item , h = self , app = self ) <EOL> @ property <EOL> def last_request_id ( self ) : <EOL> return self . _last_request_id <EOL> class ResponseError ( ValueError ) : <EOL> """<STR_LIT>""" </s>
