import pandas
from ppc_robot_lib import tasks
from ppc_robot_lib.google_ads.exceptions import handle_google_ads_errors
from ppc_robot_lib.google_ads.query import Query, Op, Condition
from ppc_robot_lib.google_ads.tasks import get_current_google_ads_facade
from ppc_robot_lib.utils.iter import chunks
OP_IN_LIMIT = 20_000
[docs]
def download_ads_report(query: Query = None) -> pandas.DataFrame:
"""
Downloads Google Ads resources. Records are stored in a table.
Example::
>>> from ppc_robot_lib.google_ads.query import Query, Condition, LastNDaysRange, OrderBy, Op, Sort
>>> from ppc_robot_lib.reporting.input.ads import download_ads_report
>>> gaql_query = Query(
... select=['campaign.name', 'metrics.clicks'],
... from_resource='campaign',
... during=LastNDaysRange(7),
... where=[
... Condition('metrics.clicks', Op.GREATER_THAN, 10)
... ],
... order_by=[OrderBy('metrics.clicks', Sort.ASCENDING)],
... limit=100
... )
>>> download_ads_report(gaql_query)
You can use `Interactive GAQL Builder <https://interactive-gaql-builder.appspot.com/>`_ to build your queries.
:param query: Query to be executed
:return: DataFrame with filled data
"""
task_ctx = tasks.get_context()
google_ads_facade = get_current_google_ads_facade()
customer_id = task_ctx.client_id
with handle_google_ads_errors(
google_ads_facade.client,
'Downloading Google Ads client (%(customer_id)s) %(report_name)s report.',
report_name=query._from,
customer_id=customer_id,
):
records = google_ads_facade.download_report(query, customer_id)
return pandas.DataFrame(data=records, columns=query.select)
def download_chunked_ads_report(query: Query = None) -> pandas.DataFrame:
"""checks if Op.IN is in and over limit right side of condition field divides to chunks"""
chunk_condition = None
where: list[Condition] = []
for condition in query.where:
if condition.operation is Op.IN and len(condition.value) > OP_IN_LIMIT:
if chunk_condition:
raise ValueError(f'Two Op.IN right value of Query are over limit {OP_IN_LIMIT}.')
else:
chunk_condition = condition
else:
where.append(condition)
if chunk_condition:
reports: list[pandas.DataFrame] = []
for chunk_value in chunks(chunk_condition.value, OP_IN_LIMIT):
new_condition = Condition(chunk_condition.column, chunk_condition.operation, chunk_value)
reports.append(download_ads_report(query.create(where=[*where, new_condition])))
else:
return download_ads_report(query)
return pandas.concat(reports, ignore_index=True)