# Copyright 2014 Facebook, Inc. # You are hereby granted a non-exclusive, worldwide, royalty-free license to # use, copy, modify, and distribute this software in source code or binary # form for use in connection with the web services and APIs provided by # Facebook. # As with any software that integrates with the Facebook platform, your use # of this software is subject to the Facebook Developer Principles and # Policies [http://developers.facebook.com/policy/]. This copyright notice # shall be included in all copies or substantial portions of the software. # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL # THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING # FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER # DEALINGS IN THE SOFTWARE. from facebook_business.session import FacebookSession from facebook_business import apiconfig from facebook_business.exceptions import ( FacebookRequestError, FacebookBadObjectError, FacebookUnavailablePropertyException, FacebookBadParameterError, ) from facebook_business.utils import api_utils from facebook_business.utils import urls from contextlib import contextmanager import copy from six.moves import http_client import os import json import six import re try: # Since python 3 from six.moves import collections_abc except ImportError: # Won't work after python 3.8 import collections as collections_abc from facebook_business.adobjects.objectparser import ObjectParser from facebook_business.typechecker import TypeChecker """ api module contains classes that make http requests to Facebook's graph API. """ class FacebookResponse(object): """Encapsulates an http response from Facebook's Graph API.""" def __init__(self, body=None, http_status=None, headers=None, call=None): """Initializes the object's internal data. Args: body (optional): The response body as text. http_status (optional): The http status code. headers (optional): The http headers. call (optional): The original call that was made. """ self._body = body self._http_status = http_status self._headers = headers or {} self._call = call def body(self): """Returns the response body.""" return self._body def json(self): """Returns the response body -- in json if possible.""" try: return json.loads(self._body) except (TypeError, ValueError): return self._body def headers(self): """Return the response headers.""" return self._headers def etag(self): """Returns the ETag header value if it exists.""" return self._headers.get('ETag') def status(self): """Returns the http status code of the response.""" return self._http_status def is_success(self): """Returns boolean indicating if the call was successful.""" json_body = self.json() if isinstance(json_body, collections_abc.Mapping) and 'error' in json_body: # Is a dictionary, has error in it return False elif bool(json_body): # Has body and no error if 'success' in json_body: return json_body['success'] # API can retuen a success 200 when service unavailable occurs return 'Service Unavailable' not in json_body elif self._http_status == http_client.NOT_MODIFIED: # ETAG Hit return True elif self._http_status == http_client.OK: # HTTP Okay return True else: # Something else return False def is_failure(self): """Returns boolean indicating if the call failed.""" return not self.is_success() def error(self): """ Returns a FacebookRequestError (located in the exceptions module) with an appropriate debug message. """ if self.is_failure(): return FacebookRequestError( "Call was not successful", self._call, self.status(), self.headers(), self.body(), ) else: return None class FacebookAdsApi(object): """Encapsulates session attributes and methods to make API calls. Attributes: SDK_VERSION (class): indicating sdk version. HTTP_METHOD_GET (class): HTTP GET method name. HTTP_METHOD_POST (class): HTTP POST method name HTTP_METHOD_DELETE (class): HTTP DELETE method name HTTP_DEFAULT_HEADERS (class): Default HTTP headers for requests made by this sdk. """ SDK_VERSION = apiconfig.ads_api_config['SDK_VERSION'] API_VERSION = apiconfig.ads_api_config['API_VERSION'] HTTP_METHOD_GET = 'GET' HTTP_METHOD_POST = 'POST' HTTP_METHOD_DELETE = 'DELETE' HTTP_DEFAULT_HEADERS = { 'User-Agent': "fbbizsdk-python-%s" % SDK_VERSION, } _default_api = None _default_account_id = None def __init__(self, session, api_version=None, enable_debug_logger=False): """Initializes the api instance. Args: session: FacebookSession object that contains a requests interface and attribute GRAPH (the Facebook GRAPH API URL). api_version: API version """ self._session = session self._num_requests_succeeded = 0 self._num_requests_attempted = 0 self._api_version = api_version or self.API_VERSION self._enable_debug_logger = enable_debug_logger def get_num_requests_attempted(self): """Returns the number of calls attempted.""" return self._num_requests_attempted def get_num_requests_succeeded(self): """Returns the number of calls that succeeded.""" return self._num_requests_succeeded @classmethod def init( cls, app_id=None, app_secret=None, access_token=None, account_id=None, api_version=None, proxies=None, timeout=None, debug=False, crash_log=True, ): session = FacebookSession(app_id, app_secret, access_token, proxies, timeout) api = cls(session, api_version, enable_debug_logger=debug) cls.set_default_api(api) if account_id: cls.set_default_account_id(account_id) if crash_log: from facebook_business.crashreporter import CrashReporter if debug: CrashReporter.enableLogging() CrashReporter.enable() return api @classmethod def set_default_api(cls, api_instance): """Sets the default api instance. When making calls to the api, objects will revert to using the default api if one is not specified when initializing the objects. Args: api_instance: The instance which to set as default. """ cls._default_api = api_instance @classmethod def get_default_api(cls): """Returns the default api instance.""" return cls._default_api @classmethod def set_default_account_id(cls, account_id): account_id = str(account_id) if account_id.find('act_') == -1: raise ValueError( "Account ID provided in FacebookAdsApi.set_default_account_id " "expects a string that begins with 'act_'", ) cls._default_account_id = account_id @classmethod def get_default_account_id(cls): return cls._default_account_id def call( self, method, path, params=None, headers=None, files=None, url_override=None, api_version=None, ): """Makes an API call. Args: method: The HTTP method name (e.g. 'GET'). path: A tuple of path tokens or a full URL string. A tuple will be translated to a url as follows: graph_url/tuple[0]/tuple[1]... It will be assumed that if the path is not a string, it will be iterable. params (optional): A mapping of request parameters where a key is the parameter name and its value is a string or an object which can be JSON-encoded. headers (optional): A mapping of request headers where a key is the header name and its value is the header value. files (optional): An optional mapping of file names to binary open file objects. These files will be attached to the request. Returns: A FacebookResponse object containing the response body, headers, http status, and summary of the call that was made. Raises: FacebookResponse.error() if the request failed. """ if not params: params = {} if not headers: headers = {} if not files: files = {} api_version = api_version or self._api_version if api_version and not re.search('v[0-9]+\.[0-9]+', api_version): raise FacebookBadObjectError( 'Please provide the API version in the following format: %s' % self.API_VERSION, ) self._num_requests_attempted += 1 if not isinstance(path, six.string_types): # Path is not a full path path = "/".join(( url_override or self._session.GRAPH, api_version, '/'.join(map(str, path)), )) # Include api headers in http request headers = headers.copy() headers.update(FacebookAdsApi.HTTP_DEFAULT_HEADERS) if params: params = _top_level_param_json_encode(params) # Get request response and encapsulate it in a FacebookResponse if method in ('GET', 'DELETE'): response = self._session.requests.request( method, path, params=params, headers=headers, files=files, timeout=self._session.timeout ) else: response = self._session.requests.request( method, path, data=params, headers=headers, files=files, timeout=self._session.timeout ) if self._enable_debug_logger: import curlify print(curlify.to_curl(response.request)) fb_response = FacebookResponse( body=response.text, headers=response.headers, http_status=response.status_code, call={ 'method': method, 'path': path, 'params': params, 'headers': headers, 'files': files, }, ) if fb_response.is_failure(): raise fb_response.error() self._num_requests_succeeded += 1 return fb_response def new_batch(self): """ Returns a new FacebookAdsApiBatch, which when executed will go through this api. """ return FacebookAdsApiBatch(api=self) class FacebookAdsApiBatch(object): """ Exposes methods to build a sequence of calls which can be executed with a single http request. Note: Individual exceptions won't be thrown for each call that fails. The success and failure callback functions corresponding to a call should handle its success or failure. """ def __init__(self, api, success=None, failure=None): self._api = api self._files = [] self._batch = [] self._success_callbacks = [] self._failure_callbacks = [] if success is not None: self._success_callbacks.append(success) if failure is not None: self._failure_callbacks.append(failure) self._requests = [] def __len__(self): return len(self._batch) def add( self, method, relative_path, params=None, headers=None, files=None, success=None, failure=None, request=None, ): """Adds a call to the batch. Args: method: The HTTP method name (e.g. 'GET'). relative_path: A tuple of path tokens or a relative URL string. A tuple will be translated to a url as follows: //... It will be assumed that if the path is not a string, it will be iterable. params (optional): A mapping of request parameters where a key is the parameter name and its value is a string or an object which can be JSON-encoded. headers (optional): A mapping of request headers where a key is the header name and its value is the header value. files (optional): An optional mapping of file names to binary open file objects. These files will be attached to the request. success (optional): A callback function which will be called with the FacebookResponse of this call if the call succeeded. failure (optional): A callback function which will be called with the FacebookResponse of this call if the call failed. request (optional): The APIRequest object Returns: A dictionary describing the call. """ if not isinstance(relative_path, six.string_types): relative_url = '/'.join(relative_path) else: relative_url = relative_path call = { 'method': method, 'relative_url': relative_url, } if params: params = _top_level_param_json_encode(params) keyvals = ['%s=%s' % (key, urls.quote_with_encoding(value)) for key, value in params.items()] if method == 'GET': call['relative_url'] += '?' + '&'.join(keyvals) else: call['body'] = '&'.join(keyvals) if files: call['attached_files'] = ','.join(files.keys()) if headers: call['headers'] = [] for header in headers: batch_formatted_header = {} batch_formatted_header['name'] = header batch_formatted_header['value'] = headers[header] call['headers'].append(batch_formatted_header) self._batch.append(call) self._files.append(files) self._success_callbacks.append(success) self._failure_callbacks.append(failure) self._requests.append(request) return call def add_request( self, request, success=None, failure=None, ): """Interface to add a APIRequest to the batch. Args: request: The APIRequest object to add success (optional): A callback function which will be called with the FacebookResponse of this call if the call succeeded. failure (optional): A callback function which will be called with the FacebookResponse of this call if the call failed. Returns: A dictionary describing the call. """ updated_params = copy.deepcopy(request._params) if request._fields: updated_params['fields'] = ','.join(request._fields) return self.add( method=request._method, relative_path=request._path, params=updated_params, files=request._file_params, success=success, failure=failure, request=request, ) def execute(self): """Makes a batch call to the api associated with this object. For each individual call response, calls the success or failure callback function if they were specified. Note: Does not explicitly raise exceptions. Individual exceptions won't be thrown for each call that fails. The success and failure callback functions corresponding to a call should handle its success or failure. Returns: If some of the calls have failed, returns a new FacebookAdsApiBatch object with those calls. Otherwise, returns None. """ if not self._batch: return None method = 'POST' path = tuple() params = {'batch': self._batch} files = {} for call_files in self._files: if call_files: files.update(call_files) fb_response = self._api.call( method, path, params=params, files=files, ) responses = fb_response.json() retry_indices = [] for index, response in enumerate(responses): if response: body = response.get('body') code = response.get('code') headers = response.get('headers') inner_fb_response = FacebookResponse( body=body, headers=headers, http_status=code, call=self._batch[index], ) if inner_fb_response.is_success(): if self._success_callbacks[index]: self._success_callbacks[index](inner_fb_response) elif self._failure_callbacks[index]: self._failure_callbacks[index](inner_fb_response) else: retry_indices.append(index) if retry_indices: new_batch = self.__class__(self._api) new_batch._files = [self._files[index] for index in retry_indices] new_batch._batch = [self._batch[index] for index in retry_indices] new_batch._success_callbacks = [self._success_callbacks[index] for index in retry_indices] new_batch._failure_callbacks = [self._failure_callbacks[index] for index in retry_indices] return new_batch else: return None class FacebookRequest: """ Represents an API request """ def __init__( self, node_id, method, endpoint, api=None, param_checker=TypeChecker({}, {}), target_class=None, api_type=None, allow_file_upload=False, response_parser=None, include_summary=True, api_version=None, ): """ Args: node_id: The node id to perform the api call. method: The HTTP method of the call. endpoint: The edge of the api call. api (optional): The FacebookAdsApi object. param_checker (optional): Parameter checker. target_class (optional): The return class of the api call. api_type (optional): NODE or EDGE type of the call. allow_file_upload (optional): Whether the call allows upload. response_parser (optional): An ObjectParser to parse response. include_summary (optional): Include "summary". api_version (optional): API version. """ self._api = api or FacebookAdsApi.get_default_api() self._node_id = node_id self._method = method self._endpoint = endpoint.replace('/', '') self._path = (node_id, endpoint.replace('/', '')) self._param_checker = param_checker self._target_class = target_class self._api_type = api_type self._allow_file_upload = allow_file_upload self._response_parser = response_parser self._include_summary = include_summary self._api_version = api_version self._params = {} self._fields = [] self._file_params = {} self._file_counter = 0 self._accepted_fields = [] if target_class is not None: self._accepted_fields = target_class.Field.__dict__.values() def add_file(self, file_path): if not self._allow_file_upload: api_utils.warning('Endpoint ' + self._endpoint + ' cannot upload files') file_key = 'source' + str(self._file_counter) if os.path.isfile(file_path): self._file_params[file_key] = file_path self._file_counter += 1 else: raise FacebookBadParameterError( 'Cannot find file ' + file_path + '!', ) return self def add_files(self, files): if files is None: return self for file_path in files: self.add_file(file_path) return self def add_field(self, field): if field not in self._fields: self._fields.append(field) if field not in self._accepted_fields: api_utils.warning(self._endpoint + ' does not allow field ' + field) return self def add_fields(self, fields): if fields is None: return self for field in fields: self.add_field(field) return self def add_param(self, key, value): if not self._param_checker.is_valid_pair(key, value): api_utils.warning('value of ' + key + ' might not be compatible. ' + ' Expect ' + self._param_checker.get_type(key) + '; ' + ' got ' + str(type(value))) if self._param_checker.is_file_param(key): self._file_params[key] = value else: self._params[key] = self._extract_value(value) return self def add_params(self, params): if params is None: return self for key in params.keys(): self.add_param(key, params[key]) return self def get_fields(self): return list(self._fields) def get_params(self): return copy.deepcopy(self._params) def execute(self): params = copy.deepcopy(self._params) if self._api_type == "EDGE" and self._method == "GET": cursor = Cursor( target_objects_class=self._target_class, params=params, fields=self._fields, include_summary=self._include_summary, api=self._api, node_id=self._node_id, endpoint=self._endpoint, ) cursor.load_next_page() return cursor if self._fields: params['fields'] = ','.join(self._fields) with open_files(self._file_params) as files: response = self._api.call( method=self._method, path=(self._path), params=params, files=files, api_version=self._api_version, ) if response.error(): raise response.error() if self._response_parser: return self._response_parser.parse_single(response.json()) else: return response def add_to_batch(self, batch, success=None, failure=None): batch.add_request(self, success, failure) def _extract_value(self, value): if hasattr(value, 'export_all_data'): return value.export_all_data() elif isinstance(value, list): return [self._extract_value(item) for item in value] elif isinstance(value, dict): return dict((self._extract_value(k), self._extract_value(v)) for (k, v) in value.items()) else: return value class Cursor(object): """Cursor is an cursor over an object's connections. Previously called EdgeIterator. Examples: >>> me = AdAccountUser('me') >>> my_accounts = [act for act in Cursor(me, AdAccount)] >>> my_accounts [, ] """ def __init__( self, source_object=None, target_objects_class=None, fields=None, params=None, include_summary=True, api=None, node_id=None, endpoint=None, object_parser=None ): """ Initializes an cursor over the objects to which there is an edge from source_object. To initialize, you'll need to provide either (source_object and target_objects_class) or (api, node_id, endpoint, and object_parser) Args: source_object: An AbstractObject instance from which to inspect an edge. This object should have an id. target_objects_class: Objects traverersed over will be initialized with this AbstractObject class. fields (optional): A list of fields of target_objects_class to automatically read in. params (optional): A mapping of request parameters where a key is the parameter name and its value is a string or an object which can be JSON-encoded. include_summary (optional): Include summary. api (optional): FacebookAdsApi object. node_id (optional): The ID of calling node. endpoint (optional): The edge name. object_parser (optional): The ObjectParser to parse response. """ self.params = dict(params or {}) target_objects_class._assign_fields_to_params(fields, self.params) self._source_object = source_object self._target_objects_class = target_objects_class self._node_id = node_id or source_object.get_id_assured() self._endpoint = endpoint or target_objects_class.get_endpoint() self._api = api or source_object.get_api() self._path = ( self._node_id, self._endpoint, ) self._queue = [] self._headers = [] self._finished_iteration = False self._total_count = None self._summary = None self._include_summary = include_summary or 'default_summary' in self.params self._object_parser = object_parser or ObjectParser( api=self._api, target_class=self._target_objects_class, ) def __repr__(self): return str(self._queue) def __len__(self): return len(self._queue) def __iter__(self): return self def __next__(self): # Load next page at end. # If load_next_page returns False, raise StopIteration exception if not self._queue and not self.load_next_page(): raise StopIteration() return self._queue.pop(0) # Python 2 compatibility. next = __next__ def __getitem__(self, index): return self._queue[index] def headers(self): return self._headers def total(self): if self._total_count is None: raise FacebookUnavailablePropertyException( "Couldn't retrieve the object total count for that type " "of request.", ) return self._total_count def summary(self): if self._summary is None or not isinstance(self._summary, dict): raise FacebookUnavailablePropertyException( "Couldn't retrieve the object summary for that type " "of request.", ) return " %s" % ( json.dumps( self._summary, sort_keys=True, indent=4, separators=(',', ': '), ), ) def load_next_page(self): """Queries server for more nodes and loads them into the internal queue. Returns: True if successful, else False. """ if self._finished_iteration: return False if ( self._include_summary and 'default_summary' not in self.params and 'summary' not in self.params ): self.params['summary'] = True response_obj = self._api.call( 'GET', self._path, params=self.params, ) response = response_obj.json() self._headers = response_obj.headers() if 'paging' in response and 'next' in response['paging']: self._path = response['paging']['next'] self.params = {} else: # Indicate if this was the last page self._finished_iteration = True if ( self._include_summary and 'summary' in response and 'total_count' in response['summary'] ): self._total_count = response['summary']['total_count'] if self._include_summary and 'summary' in response: self._summary = response['summary'] self._queue = self.build_objects_from_response(response) return len(self._queue) > 0 def get_one(self): for obj in self: return obj return None def build_objects_from_response(self, response): return self._object_parser.parse_multiple(response) @contextmanager def open_files(files): opened_files = {} for key, path in files.items(): opened_files.update({key: open(path, 'rb')}) yield opened_files for file in opened_files.values(): file.close() def _top_level_param_json_encode(params): params = params.copy() for param, value in params.items(): if ( isinstance(value, (collections_abc.Mapping, collections_abc.Sequence, bool)) and not isinstance(value, six.string_types) ): params[param] = json.dumps( value, sort_keys=True, separators=(',', ':'), ) else: params[param] = value return params