#!/usr/bin/python # # Copyright 2009 Google Inc. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # """Methods to access InfoService service.""" __author__ = 'api.sgrinberg@gmail.com (Stan Grinberg)' import datetime from aw_api import SOAPPY from aw_api import ZSI from aw_api import SanityCheck as glob_sanity_check from aw_api import Utils from aw_api.Errors import ApiVersionNotSupportedError from aw_api.Errors import ValidationError from aw_api.WebService import WebService class InfoService(object): """Wrapper for InfoService. The Info Service allows you to obtain some basic information about your API usage. """ def __init__(self, headers, config, op_config, lock, logger): """Inits InfoService. Args: headers: dict dictionary object with populated authentication credentials. config: dict dictionary object with populated configuration values. op_config: dict dictionary object with additional configuration values for this operation. lock: thread.lock the thread lock. logger: Logger the instance of Logger """ url = [op_config['server'], 'api/adwords', op_config['version'], self.__class__.__name__] if glob_sanity_check.IsNewApi(op_config['version']): url.insert(2, 'info') if config['access']: url.insert(len(url) - 1, config['access']) self.__service = WebService(headers, config, op_config, '/'.join(url), lock, logger) self.__config = config self.__op_config = op_config if self.__config['soap_lib'] == SOAPPY: from aw_api.soappy_toolkit import SanityCheck self.__web_services = None elif self.__config['soap_lib'] == ZSI: from aw_api import API_VERSIONS from aw_api.zsi_toolkit import SanityCheck if op_config['version'] in API_VERSIONS: module = '%s_services' % self.__class__.__name__ try: web_services = __import__('aw_api.zsi_toolkit.%s.%s' % (op_config['version'], module), globals(), locals(), ['']) except ImportError, e: # If one of library's required modules is missing, re raise exception. if str(e).find(module) < 0: raise ImportError(e) msg = ('The version \'%s\' is not compatible with \'%s\'.' % (op_config['version'], self.__class__.__name__)) raise ValidationError(msg) else: msg = 'Invalid API version, not one of %s.' % str(list(API_VERSIONS)) raise ValidationError(msg) self.__web_services = web_services self.__loc = eval('web_services.%sLocator()' % self.__class__.__name__) self.__sanity_check = SanityCheck def GetFreeUsageQuotaThisMonth(self): """Retrieve the number of free API units for this entire month. Free units that can be used by the developer token being used to make this call. Returns: tuple response from the API method. """ method_name = 'getFreeUsageQuotaThisMonth' if self.__config['soap_lib'] == SOAPPY: return self.__service.CallMethod(method_name, ()) elif self.__config['soap_lib'] == ZSI: web_services = self.__web_services request = eval('web_services.%sRequest()' % method_name) return self.__service.CallMethod(method_name, (), 'Info', self.__loc, request) def GetMethodCost(self, service, method, date): """Retrieve the method cost. Method cost in API units per operation, of the given method on a specific date for the developer token being used to make this call. Args: service: str name of the service containing the method. method: str method whose quota cost is being queried. date: str date for which to retrieve the cost of the method. Ex: service = 'InfoService' method = 'getMethodCost' date = '2009-01-01' Returns: tuple response from the API method. """ glob_sanity_check.ValidateTypes(((service, (str, unicode)), (method, (str, unicode)), (date, (str, unicode)))) method_name = 'getMethodCost' if self.__config['soap_lib'] == SOAPPY: return self.__service.CallMethod(method_name, (service, method, date)) elif self.__config['soap_lib'] == ZSI: web_services = self.__web_services request = eval('web_services.%sRequest()' % method_name) return self.__service.CallMethod(method_name, (({'service': service}, {'method': method}, {'date': date})), 'Info', self.__loc, request) def GetOperationCount(self, start_date, end_date): """Retrieve the number of operations. Operations recorded for the developer token being used to make this call over the given date range. Args: start_date: str beginning of the date range, inclusive end_date: str end of the date range, inclusive Ex: start_date = '2009-01-01' end_date = '2009-01-31' Returns: tuple response from the API method. """ glob_sanity_check.ValidateTypes(((start_date, (str, unicode)), (end_date, (str, unicode)))) method_name = 'getOperationCount' if self.__config['soap_lib'] == SOAPPY: return self.__service.CallMethod(method_name, (start_date, end_date)) elif self.__config['soap_lib'] == ZSI: web_services = self.__web_services request = eval('web_services.%sRequest()' % method_name) return self.__service.CallMethod(method_name, (({'startDate': start_date}, {'endDate': end_date})), 'Info', self.__loc, request) def GetUnitCount(self, start_date, end_date): """Retrieve the number of API units. Units recorded for the developer token being used to make this call over the given date range. Args: start_date: str beginning of the date range, inclusive end_date: str end of the date range, inclusive Ex: start_date = '2009-01-01' end_date = '2009-01-31' Returns: tuple response from the API method. """ glob_sanity_check.ValidateTypes(((start_date, (str, unicode)), (end_date, (str, unicode)))) method_name = 'getUnitCount' if self.__config['soap_lib'] == SOAPPY: return self.__service.CallMethod(method_name, (start_date, end_date)) elif self.__config['soap_lib'] == ZSI: web_services = self.__web_services request = eval('web_services.%sRequest()' % method_name) return self.__service.CallMethod(method_name, (({'startDate': start_date}, {'endDate': end_date})), 'Info', self.__loc, request) def GetUnitCountForClients(self, client_emails, start_date, end_date): """Retrieve the number of API units recorded for clients. Units recorded for a subset of clients over the given date range for the developer token being used to make this call. Args: client_emails: list login emails used to identify the clients to retrieve API units for. start_date: str beginning of the date range, inclusive. end_date: str end of the date range, inclusive. Ex: client_emails = ['johndoe@example.com'] start_date = '2009-01-01' end_date = '2009-01-31' Returns: tuple response from the API method. """ glob_sanity_check.ValidateTypes(((client_emails, list), (start_date, (str, unicode)), (end_date, (str, unicode)))) for item in client_emails: glob_sanity_check.ValidateTypes(((item, (str, unicode)),)) method_name = 'getUnitCountForClients' if self.__config['soap_lib'] == SOAPPY: return self.__service.CallMethod(method_name, (client_emails, start_date, end_date)) elif self.__config['soap_lib'] == ZSI: web_services = self.__web_services request = eval('web_services.%sRequest()' % method_name) return self.__service.CallMethod(method_name, (({'clientEmails': client_emails}, {'startDate': start_date}, {'endDate': end_date})), 'Info', self.__loc, request) def GetUnitCountForMethod(self, service, method, start_date, end_date): """Retrieve the number of API units recorded per method. Units recorded for the developer token being used to make this call over the given date range for a specific method. Args: service: str name of the service containing the method. method: str method whose quota usage is being queried. start_date: str beginning of the date range, inclusive. end_date: str end of the date range, inclusive. Ex: service = 'InfoService' method = 'getUnitCountForMethod' start_date = '2009-01-01' end_date = '2009-01-31' Returns: tuple response from the API method. """ glob_sanity_check.ValidateTypes(((service, (str, unicode)), (method, (str, unicode)), (start_date, (str, unicode)), (end_date, (str, unicode)))) method_name = 'getUnitCountForMethod' if self.__config['soap_lib'] == SOAPPY: return self.__service.CallMethod(method_name, (service, method, start_date, end_date)) elif self.__config['soap_lib'] == ZSI: web_services = self.__web_services request = eval('web_services.%sRequest()' % method_name) return self.__service.CallMethod(method_name, (({'service': service}, {'method': method}, {'startDate': start_date}, {'endDate': end_date})), 'Info', self.__loc, request) def GetUsageQuotaThisMonth(self): """Retrieve the total number of API units for this entire month. Units that can be used by the developer token being used to make this call. Returns: tuple response from the API method. """ method_name = 'getUsageQuotaThisMonth' if self.__config['soap_lib'] == SOAPPY: return self.__service.CallMethod(method_name, ()) elif self.__config['soap_lib'] == ZSI: web_services = self.__web_services request = eval('web_services.%sRequest()' % method_name) return self.__service.CallMethod(method_name, (), 'Info', self.__loc, request) def GetUnitSummary(self): """Retrieve the current status of API units. The results are Similar to the numbers shown in UI. Consumes 3 API units. Returns: tuple collection of API units stats. """ total_limit = self.GetUsageQuotaThisMonth()[0] free_quota_limit = self.GetFreeUsageQuotaThisMonth()[0] start_date = datetime.datetime.now().strftime('%Y-%m-01T00:00:00') end_date = datetime.datetime.now().strftime('%Y-%m-%dT00:00:00') unit_count = self.GetUnitCount(start_date, end_date)[0] if long(unit_count) > long(free_quota_limit): free_quota_used = free_quota_limit else: free_quota_used = unit_count stats = { 'free_units_used': free_quota_used, 'total_units_used': unit_count, 'free_units_remaining': long(free_quota_limit) - long(free_quota_used), 'total_units': total_limit } return (stats,) def ShowUnitSummary(self): """Print the current status of API units. The results are similar to the numbers shown in UI. """ stats = self.GetUnitSummary()[0] print 'Free API units used: \'%s\'' % stats['free_units_used'] print 'Total API units used: \'%s\'' % stats['total_units_used'] print 'Free API units remaining: \'%s\'' % stats['free_units_remaining'] print 'System-defined quota cap: \'%s\'' % stats['total_units'] def GetTotalUnitUsed(self): """Retrieve the total number of API units used. Units used from beggining of time to now. Returns: tuple total number of API units used. """ start_date = '2004-01-01T00:00:00' end_date = datetime.datetime.now().strftime('%Y-%m-%dT00:00:00') unit_count = self.GetUnitCount(start_date, end_date)[0] return (unit_count,) def GetUnitDetails(self, start_date='%Y-%m-01', end_date='%Y-%m-%d'): """Retrieve details of the API units usage for given date range. Defaults to current month. Consumes 66 API units. Args: [optional] start_date: str beginning of the date range, inclusive. end_date: str end of the date range, inclusive. Ex: start_date = '2009-01-01' end_date = '2009-01-10' Returns: tuple detailed API usage. """ glob_sanity_check.ValidateTypes(((start_date, (str, unicode)), (end_date, (str, unicode)))) date = start_date start_date = datetime.datetime.now().strftime('%sT00:00:00' % start_date) end_date = datetime.datetime.now().strftime('%sT00:00:00' % end_date) ops_rates = Utils.GetOpsRates() usage = [] parts = [] for parts in ops_rates: # TODO(api.sgrinberg): Add support for v2009 operations. if parts[0] == 'v13': unit_count = self.GetUnitCountForMethod( parts[1], parts[2], start_date, end_date)[0] if long(unit_count) > 0: method_usage = { 'date': date, 'service': parts[1], 'method': parts[2], 'units': unit_count } usage.append(method_usage) return (usage,) def GetUnitDetailsDaily(self, date_day=None): """Retrieve details of the API units usage per day for current month. Consumes up to 2046 API units. Args: date_day: str number of days to look up. Ex: date_day = '1' Returns: tuple detailed API usage per day. """ if date_day is None: date_day = int(datetime.datetime.now().strftime('%d')) else: glob_sanity_check.ValidateTypes(((date_day, (str, unicode)),)) date_day = int(date_day) usage = [] for day in range(1, date_day + 1): start_date = datetime.datetime.now().strftime('%%Y-%%m-%s' % day) end_date = start_date day_usage = self.GetUnitDetails(start_date, end_date)[0] if day_usage: usage.append(day_usage) return (usage,) def ShowUnitDetails(self, start_date=None, end_date=None): """Print details of the API units usage for given date range. Defaults to current month. Args: [optional] start_date: str beginning of the date range, inclusive. end_date: str end of the date range, inclusive. Ex: start_date = '2009-01-01' end_date = '2009-01-10' """ if start_date is None or end_date is None: usage = self.GetUnitDetails()[0] else: usage = self.GetUnitDetails(start_date, end_date)[0] if not usage: print 'No API usage recorded during this month.' else: print '%28s %28s %28s' % ('Service', 'Method', 'Units') total_units = 0 for method_usage in usage: print '%28s %28s %28s' % (method_usage['service'], method_usage['method'], method_usage['units']) total_units += long(method_usage['units']) print print 'Total API units used: %s' % total_units def ShowUnitDetailsDaily(self): """Print details of the API unis usage per day.""" usage = self.GetUnitDetailsDaily()[0] if not usage: print 'No API usage recorded during this monht.' else: print '%28s %28s %28s %28s' % ('Date', 'Service', 'Method', 'Units') total_units = 0 for method_usage in usage: print '%28s %28s %28s %28s' % (method_usage[0]['date'], method_usage[0]['service'], method_usage[0]['method'], method_usage[0]['units']) total_units += long(method_usage[0]['units']) print print 'Total API units used: %s' % total_units def Get(self, selector): """Return the API usage information. Usage information is based on the selection criteria of the selector. Args: selector: dict filter to run API usage through. Returns: tuple API usage information. """ method_name = 'getInfo' if self.__config['soap_lib'] == SOAPPY: msg = ('The \'%s\' request via %s is currently not supported for ' 'use with SOAPpy toolkit.' % (Utils.GetCurrentFuncName(), self.__op_config['version'])) raise ApiVersionNotSupportedError(msg) elif self.__config['soap_lib'] == ZSI: web_services = self.__web_services self.__sanity_check.ValidateSelector(selector, web_services) request = eval('web_services.%sRequest()' % method_name) return self.__service.CallMethod(method_name, (({'selector': selector},)), 'Info', self.__loc, request)