Package upwork
Top-level package for python-upwork.
Expand source code
"""Top-level package for python-upwork."""
from upwork.config import Config
from upwork.client import Client
from . import routers
__author__ = """Maksym Novozhylov"""
__email__ = "mnovozhilov@upwork.com"
__version__ = "3.1.0"
__all__ = ("Config", "Client", "routers")
Sub-modules
upwork.client
upwork.config
upwork.routers
-
routers
upwork.upwork
-
Main module.
Classes
class Client (config)
-
API client for OAuth2 authorization
Parameters: :config: An instance of upwork.Config class, which contains the configuration keys and tokens
Expand source code
class Client(object): """API client for OAuth2 authorization *Parameters:* :config: An instance of upwork.Config class, which contains the configuration keys and tokens """ __data_format = "json" __overload_var = "http_method" __uri_auth = "/ab/account-security/oauth2/authorize" __uri_rtoken = "/v3/oauth2/token" __uri_atoken = "/v3/oauth2/token" epoint = upwork.DEFAULT_EPOINT def __init__(self, config): self.config = config self.config.tenant_id = None try: # token is known, use it self.__oauth = OAuth2Session( self.config.client_id, token=self.config.token, auto_refresh_url=full_url(self.__uri_rtoken, upwork.DEFAULT_EPOINT), auto_refresh_kwargs={ "client_id": self.config.client_id, "client_secret": self.config.client_secret, }, token_updater=self.refresh_config_from_access_token, ) except AttributeError as e: if self.config.grant_type == "client_credentials": client = BackendApplicationClient(client_id=self.config.client_id) self.__oauth = OAuth2Session( client=client ) else: # start from authorization step self.__oauth = OAuth2Session( self.config.client_id, redirect_uri=self.config.redirect_uri ) def get_authorization_url(self): """Get authorization URL :param redirect_uri: (Default value = None) """ return self.__oauth.authorization_url( "{0}{1}".format(upwork.BASE_HOST, self.__uri_auth) ) def get_access_token(self, authorization_response=None): """Finish auth process and get access token :param authorization_response: """ self.config.token = self.__oauth.fetch_token( full_url(self.__uri_atoken, upwork.DEFAULT_EPOINT), authorization_response=authorization_response, client_secret=self.config.client_secret, ) return self.config.token def refresh_config_from_access_token(self, token): """Callback from OAuth2 client which will refresh config with actual data""" self.config.token = token def set_org_uid_header(self, tenant_id): """Configure X-Upwork-API-TenantId header""" self.config.tenant_id = tenant_id def get_actual_config(self): """Get actual client config""" return self.config def get(self, uri, params=None): """Execute GET request :param uri: :param params: (Default value = None) """ return self.send_request(uri, "get", params) def post(self, uri, params=None): """Execute POST request :param uri: :param params: (Default value = None) """ return self.send_request(uri, "post", params) def put(self, uri, params=None): """Execute PUT request :param uri: :param params: (Default value = None) """ return self.send_request(uri, "put", params) def delete(self, uri, params=None): """Execute DELETE request :param uri: :param params: (Default value = None) """ return self.send_request(uri, "delete", params) def send_request(self, uri, method="get", params={}): """Send request :param uri: :param method: (Default value = 'get') :param params: (Default value = {}) """ # delete does not support passing the parameters if method == "delete": params[self.__overload_var] = method url = full_url(get_uri_with_format(uri, self.epoint), self.epoint) if method == "get": r = self.__oauth.get(url, params=params) elif method == "put": headers = {"Content-type": "application/json"} r = self.__oauth.put(url, json=params, headers=headers) elif method in {"post", "delete"}: headers = {"Content-type": "application/json"} if self.epoint == "graphql" and self.config.tenant_id: headers["X-Upwork-API-TenantId"] = self.config.tenant_id r = self.__oauth.post(url, json=params, headers=headers) else: raise ValueError( 'Do not know how to handle http method "{0}"'.format(method) ) return r.json()
Class variables
var epoint
Methods
def delete(self, uri, params=None)
-
Execute DELETE request
:param uri: :param params: (Default value = None)
Expand source code
def delete(self, uri, params=None): """Execute DELETE request :param uri: :param params: (Default value = None) """ return self.send_request(uri, "delete", params)
def get(self, uri, params=None)
-
Execute GET request
:param uri: :param params: (Default value = None)
Expand source code
def get(self, uri, params=None): """Execute GET request :param uri: :param params: (Default value = None) """ return self.send_request(uri, "get", params)
def get_access_token(self, authorization_response=None)
-
Finish auth process and get access token
:param authorization_response:
Expand source code
def get_access_token(self, authorization_response=None): """Finish auth process and get access token :param authorization_response: """ self.config.token = self.__oauth.fetch_token( full_url(self.__uri_atoken, upwork.DEFAULT_EPOINT), authorization_response=authorization_response, client_secret=self.config.client_secret, ) return self.config.token
def get_actual_config(self)
-
Get actual client config
Expand source code
def get_actual_config(self): """Get actual client config""" return self.config
-
Get authorization URL
:param redirect_uri: (Default value = None)
Expand source code
def get_authorization_url(self): """Get authorization URL :param redirect_uri: (Default value = None) """ return self.__oauth.authorization_url( "{0}{1}".format(upwork.BASE_HOST, self.__uri_auth) )
def post(self, uri, params=None)
-
Execute POST request
:param uri: :param params: (Default value = None)
Expand source code
def post(self, uri, params=None): """Execute POST request :param uri: :param params: (Default value = None) """ return self.send_request(uri, "post", params)
def put(self, uri, params=None)
-
Execute PUT request
:param uri: :param params: (Default value = None)
Expand source code
def put(self, uri, params=None): """Execute PUT request :param uri: :param params: (Default value = None) """ return self.send_request(uri, "put", params)
def refresh_config_from_access_token(self, token)
-
Callback from OAuth2 client which will refresh config with actual data
Expand source code
def refresh_config_from_access_token(self, token): """Callback from OAuth2 client which will refresh config with actual data""" self.config.token = token
def send_request(self, uri, method='get', params={})
-
Send request
:param uri: :param method: (Default value = 'get') :param params: (Default value = {})
Expand source code
def send_request(self, uri, method="get", params={}): """Send request :param uri: :param method: (Default value = 'get') :param params: (Default value = {}) """ # delete does not support passing the parameters if method == "delete": params[self.__overload_var] = method url = full_url(get_uri_with_format(uri, self.epoint), self.epoint) if method == "get": r = self.__oauth.get(url, params=params) elif method == "put": headers = {"Content-type": "application/json"} r = self.__oauth.put(url, json=params, headers=headers) elif method in {"post", "delete"}: headers = {"Content-type": "application/json"} if self.epoint == "graphql" and self.config.tenant_id: headers["X-Upwork-API-TenantId"] = self.config.tenant_id r = self.__oauth.post(url, json=params, headers=headers) else: raise ValueError( 'Do not know how to handle http method "{0}"'.format(method) ) return r.json()
def set_org_uid_header(self, tenant_id)
-
Configure X-Upwork-API-TenantId header
Expand source code
def set_org_uid_header(self, tenant_id): """Configure X-Upwork-API-TenantId header""" self.config.tenant_id = tenant_id
class Config (config)
-
Configuration container
Expand source code
class Config: """Configuration container""" def __init__(self, config): self.client_id, self.client_secret = ( config["client_id"], config["client_secret"], ) if "grant_type" in config: self.grant_type = config["grant_type"] else: self.grant_type = None # Authorization Code Grant flow is used by default if "redirect_uri" in config: self.redirect_uri = config["redirect_uri"] # access-, refresh token, and expires_in/at data if "token" in config: self.token = config["token"]