from __future__ import annotations
import calendar
import datetime
from abc import ABC, abstractmethod
from datetime import date
from enum import Enum
from typing import Any
from collections.abc import Sequence
from functools import cached_property
from ppc_robot_lib.google_ads import resources, transformations
class Date:
"""
Date wrapper used in date range specifications.
"""
def __init__(self, date):
if isinstance(date, str):
self.date: datetime.date = transformations.DateTransformation.to_internal_value(date)
else:
self.date: datetime.date = date
def __str__(self):
return self.date.strftime('%Y%m%d')
def __eq__(self, other):
if isinstance(other, Date):
return self.date == other.date
else:
return False
def __ne__(self, other):
if isinstance(other, Date):
return self.date != other.date
else:
return True
def __hash__(self):
return hash(self.date)
[docs]
class Query:
"""
Represents GAQL query used to retrieve data from Google Ads API.
You can use `Interactive GAQL Builder <https://interactive-gaql-builder.appspot.com/>`_ to build your queries.
Introduction and specification can be found on Google's developer documentation:
* https://developers.google.com/google-ads/api/docs/query/grammar
"""
def __init__(
self,
select: list[str] = None,
from_resource: str = None,
where: list[Condition] = None,
during: DuringClause = None,
order_by: list[OrderBy] = None,
limit: int | None = None,
parameters: Parameter | None = None,
):
"""
:param select: List of columns to select, e.g. :code:`['campaign.id', 'segments.device', 'metrics.clicks']`
:param from_resource: Name of the resource, e.g. `campaign`
:param where: List of conditions.
:param during: During clause.
:param parameters: List of meta parameters.
:param order_by: List of order-by clauses.
:param limit: Limit the number of results.
"""
self.select = select if select is not None else []
self._from = from_resource
self._where = where if where is not None else []
self._during = during
self._order_by = order_by if order_by is not None else []
self._limit = limit
self._parameters = parameters
@property
def where(self) -> list[Condition]:
return self._where
def create(
self,
select: list[str] = None,
from_resource: str = None,
where: list[Condition] = None,
during: DuringClause = None,
order_by: list[OrderBy] = None,
limit: int | None = None,
parameters: Parameter | None = None,
):
return Query(
select if select else self.select,
from_resource if from_resource else self._from,
where if where else self._where,
during if during else self._during,
order_by if order_by else self._order_by,
limit if limit else self._limit,
parameters if parameters else self._parameters,
)
[docs]
def to_gaql(self) -> str:
"""
Constructs query in GAQL.
:return: Query representation as string.
"""
conditions: list[str] = []
order_list: list[str] = []
query: list[str] = ['SELECT', ', '.join(self.select), 'FROM', self._from]
if self._during:
conditions.append(self._during.to_gaql())
for where_clause in self._where:
conditions.append(where_clause.to_gaql())
if conditions:
query.append(f'WHERE {" AND ".join(conditions)}')
for order_by_item in self._order_by:
order_list.append(order_by_item.to_gaql())
if self._order_by:
query.append(f'ORDER BY {", ".join(order_list)}')
if self._limit:
query.append(f'LIMIT {self._limit}')
if self._parameters:
query.append(f'PARAMETERS {self._parameters.to_gaql()}')
return ' '.join(query)
def __eq__(self, other):
if isinstance(other, Query):
return (
self.select == other.select
and self._during == other._during
and self._where == other._where
and self._order_by == other._order_by
and self._limit == other._limit
and self._parameters == other._parameters
)
return False
[docs]
class Sort(Enum):
"""
Sort order.
"""
ASCENDING = 'ASC'
"""Ascending."""
DESCENDING = 'DESC'
"""Descending."""
def __str__(self):
return self.value
[docs]
class OrderBy:
"""
Single ORDER BY clause.
"""
def __init__(self, column: str, sort: Sort = Sort.ASCENDING):
"""
:param column: Column to sort by.
:param sort: Sorting direction.
"""
self._column = column
self._sort = sort
def __repr__(self):
return f'OrderBy(column={self._column}, sort={self._sort})'
[docs]
def to_gaql(self) -> str:
"""
Transform to GAQL.
:return:
"""
return f'{self._column} {self._sort}'
def __eq__(self, other):
if isinstance(other, OrderBy):
return self._column == other._column and self._sort == other._sort
return False
[docs]
class DuringClause(ABC):
"""
Abstract class for during clause.
"""
[docs]
@abstractmethod
def get_title(self):
"""
:return: Title of the specified date range.
"""
pass
[docs]
@abstractmethod
def to_gaql(self) -> str:
"""
Transform to GAQL fragment: one of the reserved constants (e.g. ``LAST_30_DAYS``)
or a date pair in ``YYYYMMDD`` format.
"""
pass
[docs]
class Op(Enum):
"""
Operators used in filtering conditions.
"""
EQUALS = '='
"""Equal."""
NOT_EQUALS = '!='
"""Not equal."""
GREATER_THAN = '>'
"""Greater than."""
GREATER_THAN_EQUALS = '>='
"""Greater than or equals."""
LESS_THAN = '<'
"""Less than."""
LESS_THAN_EQUALS = '<='
"""Less than equals."""
IN = 'IN'
"""Value is equal to one of the values in list."""
NOT_IN = 'NOT IN'
"""Value is not equal to any of the values in list."""
LIKE = 'LIKE'
"""Value matches a specified pattern."""
NOT_LIKE = 'NOT LIKE'
"""Value does not match a specified pattern."""
CONTAINS_ANY = 'CONTAINS ANY'
"""Set (e.g. labels) contains at least one of the specified values."""
CONTAINS_ALL = 'CONTAINS ALL'
"""Set (e.g. labels) contains all of the specified values."""
CONTAINS_NONE = 'CONTAINS NONE'
"""Set (e.g. labels) does not contain any of the specified values."""
IS_NULL = 'IS NULL'
"""Value is null."""
IS_NOT_NULL = 'IS NOT NULL'
"""Value is not null."""
DURING = 'DURING'
"""Filter 'date' field with the predefined 'date range'."""
BETWEEN = 'BETWEEN'
"""Operator to use to select between two dates."""
REGEXP_MATCH = 'REGEXP_MATCH'
"""
Value matches a `regular expression pattern`_.
.. _regular expression pattern:
https://support.google.com/docs/answer/3098292?hl=en
"""
NOT_REGEXP_MATCH = 'NOT REGEXP_MATCH'
"""
Value does not match a `regular expression pattern`_.
.. _regular expression pattern:
https://support.google.com/docs/answer/3098292?hl=en
"""
def __str__(self):
return self.value
[docs]
class Parameter:
"""
Parameter clause to specify meta parameters for the request.
"""
def __init__(self, meta_parameter: str, value: bool):
"""
:param meta_parameter: Name of the query parameter.
:param value: Parameter value.
"""
self._meta_parameter = meta_parameter
self._value = value
def __repr__(self):
return f'Parameter(meta_parameter={self._meta_parameter}, value={str(self._value).lower()})'
def to_gaql(self):
return f'{self._meta_parameter} = {str(self._value).lower()}'
def __eq__(self, other):
if isinstance(other, Parameter):
return self._meta_parameter == other._meta_parameter and self._value == other._value
return False
[docs]
class Condition:
"""
Predicate defining a condition from GAQL.
"""
def __init__(self, column: str, op: Op, value: list | Sequence | str | int | tuple[int, int] | None | date = None):
"""
:param column: Left side of the operator: name of the column.
:param op: Operator.
:param value: Right side of the operator: value to compare to.
"""
self._column = column
self._operation = op
self._value = value
@property
def column(self):
return self._column
@property
def operation(self):
return self._operation
@property
def value(self):
return self._value
def __repr__(self):
return f'Condition(column={self._column}, operation={self._operation}, value={self._value})'
[docs]
def to_gaql(self) -> str:
"""
Transforms to GAQL query string.
:return: GAQL fragment.
"""
value = self._value
field = resources.get_field(self._column)
if field.transformation:
value = self.transform(value, field)
if self._operation is Op.BETWEEN:
if type(value[0]) is str:
return f'{self._column} {self._operation} "{value[0]}" AND "{value[1]}"'
else:
return f'{self._column} {self._operation} {value[0]} AND {value[1]}'
elif (
self._operation is Op.IN
or self._operation is Op.NOT_IN
or self._operation is Op.CONTAINS_ALL
or self._operation is Op.CONTAINS_ANY
or self._operation is Op.CONTAINS_NONE
):
list_query = []
for val in value:
if hasattr(val, 'name'):
val = val.name
if type(val) is str:
escaped_val = val.replace('"', r'\"')
list_query.append(f'"{escaped_val}"')
else:
list_query.append(f'{val}')
prepared_list_str = ', '.join(list_query)
return f'{self._column} {self._operation} ({prepared_list_str})'
else:
if type(value) is int:
return f'{self._column} {self._operation} {value}'
elif type(value) is bool:
return f'{self._column} {self._operation} {str(value).upper()}'
elif value is None:
return f'{self._column} {self._operation}'
elif type(value) is str:
escaped_val = value.replace('"', r'\"')
return f'{self._column} {self._operation} "{escaped_val}"'
elif isinstance(value, Enum):
return f'{self._column} {self._operation} "{value.name}"'
def __eq__(self, other):
if isinstance(other, Condition):
return self._column == other._column and self._operation == other._operation and self._value == other._value
return False
class DateRange(DuringClause):
"""
Fixed Date Range - both start and end times are not relative to current date.
"""
def __init__(self, start, end):
if not isinstance(start, Date):
self._start = Date(start)
else:
self._start = start
if not isinstance(end, Date):
self._end = Date(end)
else:
self._end = end
def get_title(self):
return f'{self._start} - {self._end}'
def to_gaql(self) -> str:
"""
Transforms to GAQL query string.
:return: GAQL fragment.
"""
return f'segments.date BETWEEN {self._start} AND {self._end}'
@property
def start(self):
return self._start
@property
def end(self):
return self._end
def __str__(self):
return f'{self._start},{self._end}'
def __eq__(self, other):
if isinstance(other, DateRange):
return self._start == other._start and self._end == other._end
elif isinstance(other, LastNDaysRange):
return self._start == other.start and self._end == other.end
else:
return False
def __ne__(self, other):
if isinstance(other, DateRange):
return self._start != other._start or self._end != other._end
elif isinstance(other, LastNDaysRange):
return self._start != other.start and self._end != other.end
else:
return True
def __hash__(self):
return hash((self._start, self._end))
def last_month(today: datetime.date):
prev_month = today.month - 1
if prev_month < 1:
prev_month = 12
year = today.year - 1
else:
year = today.year
_, days = calendar.monthrange(year, prev_month)
return DateRange(datetime.date(year, prev_month, 1), datetime.date(year, prev_month, days))
[docs]
class LastNDaysRange(DuringClause):
"""
Range of Last N days.
"""
def __init__(self, n_days: int, offset: int = 1):
"""
:param n_days: How many days to go back when determining start date.
:param offset: How many days to go back when determining end date. Default value of 1 equals to yesterday.
"""
self._n_days = n_days
self._offset = offset
self._date_range: DateRange | None = None
def __repr__(self):
return f'LastNDaysRange(n_days={self._n_days}, offset={self._offset})[{str(self._date_range)}] at {id(self)}'
[docs]
def get_title(self):
return f'Back to {self._n_days} days'
def _create_range(self):
"""
Create range of dates based on given start date and end date.
:return:
"""
end_date = datetime.date.today() - datetime.timedelta(days=self._offset)
start_date = datetime.date.today() - datetime.timedelta(days=self._n_days)
self._date_range = DateRange(start_date, end_date)
[docs]
def to_gaql(self) -> str:
"""
Transforms to GAQL query string.
:return: GAQL fragment.
"""
self._create_range()
return f'segments.date BETWEEN {self._date_range.start} AND {self._date_range.end}'
@property
def start(self) -> Date:
if not self._date_range:
self._create_range()
return self._date_range.start
@property
def end(self) -> Date:
if not self._date_range:
self._create_range()
return self._date_range.end
def __eq__(self, other):
if isinstance(other, LastNDaysRange):
return self._n_days == other._n_days and self._offset == other._offset
return False
class LastNMonthsRange(DuringClause):
"""
Last N months range. This range will use only full month starting from the previous month (always relative
to current date) and optionally include the current month.
"""
def __init__(self, n_months: int, include_this_month: bool = False):
self.n_months = n_months
self.include_this_month = include_this_month
self._date_range = None
def __repr__(self):
return (
f'<LastNMonthsRange(n_months={self.n_months}, include_this_month={self.include_this_month})'
f'[{str(self._date_range)}] at {id(self)}>'
)
def get_title(self):
return f'{self.n_months} months'
def _create_range(self):
today = datetime.date.today()
first_of_this_month = today.replace(day=1)
if self.include_this_month:
end_date = today
else:
end_date = first_of_this_month - datetime.timedelta(days=1)
start_date = self._month_add(first_of_this_month, -self.n_months)
self._date_range = DateRange(start_date, end_date)
@staticmethod
def _month_add(month_date: datetime.date, delta):
month, year = ((month_date.month + delta) % 12, month_date.year + (month_date.month + delta - 1) // 12)
if not month:
month = 12
d = min(month_date.day, calendar.monthrange(year, month)[1])
return month_date.replace(day=d, month=month, year=year)
def to_gaql(self) -> str:
"""
Transforms to GAQL query string.
:return: GAQL fragment.
"""
self._create_range()
return f'segments.date BETWEEN {self._date_range.start} AND {self._date_range.end}'
@property
def start(self) -> Date:
if not self._date_range:
self._create_range()
return self._date_range.start
@property
def end(self) -> Date:
if not self._date_range:
self._create_range()
return self._date_range.end
def __eq__(self, other):
if isinstance(other, LastNMonthsRange):
return self.n_months == other.n_months and self.include_this_month == other.include_this_month
return False
class RelativeRange(DuringClause):
"""
Relative date range - start and end dates are computed relative to current date.
"""
@abstractmethod
def _create_range(self) -> DateRange:
raise NotImplementedError()
@cached_property
def date_range(self) -> DateRange:
return self._create_range()
@property
def start(self) -> Date:
return self.date_range.start
@property
def end(self) -> Date:
return self.date_range.end
def to_gaql(self) -> str:
return f'segments.date BETWEEN {self.date_range.start} AND {self.date_range.end}'
def __eq__(self, other):
if isinstance(other, RelativeRange):
return self.date_range == other.date_range
elif isinstance(other, DateRange):
return self.date_range == other
else:
return False
def __repr__(self):
return f'LastRange(date_range={repr(self.date_range)})'
def __str__(self):
return str(self.date_range)
class LastNWeeksRange(RelativeRange):
"""
Last N weeks range. This range will use only full weeks starting from the previous week (always relative
to current date). Weeks begin on monday.
"""
def __init__(self, n_weeks: int):
self.n_weeks = n_weeks
super().__init__()
def get_title(self):
return f'{self.n_weeks} weeks'
def __repr__(self):
return f'<LastNWeeksRange(n_weeks={self.n_weeks})[{str(self.date_range)}] at {id(self)}>'
def _create_range(self):
today = datetime.date.today()
sunday = today - datetime.timedelta(days=today.isoweekday())
start_date = sunday - datetime.timedelta(days=(7 * self.n_weeks) - 1)
return DateRange(start_date, sunday)
class CustomDateRange(RelativeRange):
"""
:class:`DateRange` wrapper.
"""
def __init__(self, date_range: DateRange, title: str = 'Custom date range'):
self._title = title
self._date_range = date_range
def get_title(self):
return self._title
def _create_range(self):
return self._date_range
class ThisMonthToYdayRange(RelativeRange):
"""
Range for all days in the current month except today.
"""
def _create_range(self):
today = datetime.date.today()
yday = today - datetime.timedelta(days=1)
month_start = yday.replace(year=yday.year, month=yday.month, day=1)
return DateRange(month_start, yday)
def get_title(self):
return '1st of Month to Yesterday'
def __repr__(self):
return f'<ThisMonthToYdayRange()[{str(self.date_range)}] at {id(self)}>'
class LastMonthRange(RelativeRange):
"""
Take the dates from the given range and shift them to the previous month. The range should not
span over more than 365 days.
If the previous month has less days than the month given, the range is truncated to stay within that month:
e.g. [2018-03-29; 2018-03-31] becomes [2018-02-28; 2018-02-28].
"""
def __init__(self, current_range: RelativeRange | DateRange, title: str = None):
self._current = current_range
self._title = title
def get_title(self):
if self._title:
return self._title
else:
return f'{self._current.get_title()} Previous Month'
def __repr__(self):
return f'<PreviousMonthRange(current={repr(self._current)})[{str(self.date_range)}] at {id(self)}>'
def _create_range(self):
start = self._current.start.date
end = self._current.end.date
return DateRange(
self.shift_to_prev_month(start),
self.shift_to_prev_month(end),
)
@staticmethod
def shift_to_prev_month(date: datetime.date):
new_year = date.year
new_month = date.month - 1
new_day = date.day
if new_month == 0:
new_year = new_year - 1
new_month = 12
_, month_days = calendar.monthrange(new_year, new_month)
if new_day > month_days:
new_day = month_days
return datetime.date(new_year, new_month, new_day)
class LastYearRange(RelativeRange):
"""
Take the dates from the given range and shift them to the previous year. The range should not
span over more than 365 days.
"""
def __init__(self, current_range: RelativeRange | DateRange, title: str = None):
self._current = current_range
self._title = title
def get_title(self):
if self._title:
return self._title
else:
return f'{self._current.get_title()} Previous Year'
def __repr__(self):
return f'<PreviousYearRange(current={repr(self._current)})[{str(self.date_range)}] at {id(self)}>'
def _create_range(self):
start = self._current.start.date
end = self._current.end.date
if start.month != 2 or start.day != 29:
new_start = start.replace(year=start.year - 1)
else:
new_start = start.replace(year=start.year - 1, day=28)
if end.month != 2 or end.day != 29:
new_end = end.replace(year=end.year - 1)
else:
new_end = end.replace(year=end.year - 1, day=28)
return DateRange(new_start, new_end)
[docs]
class During(DuringClause):
"""
Definition of reserved during constants.
"""
TODAY = None
"""Today only."""
YESTERDAY = None
"""Yesterday only."""
LAST_7_DAYS = None
"""The last 7 days not including today."""
LAST_BUSINESS_WEEK = None
"""The 5 day business week, Monday through Friday, of the previous business week."""
THIS_MONTH = None
""" All days in the current month."""
LAST_MONTH = None
"""All days in the previous month."""
LAST_14_DAYS = None
"""The last 14 days not including today."""
LAST_30_DAYS = None
"""The last 30 days not including today."""
THIS_WEEK_SUN_TODAY = None
""" The period between the previous Sunday and the current day."""
THIS_WEEK_MON_TODAY = None
"""The period between the previous Monday and the current day."""
LAST_WEEK_SUN_SAT = None
"""The 7-day period starting with the previous Sunday."""
LAST_WEEK_MON_SUN = None
"""The 7-day period starting with the previous Monday."""
def __init__(self, reserved_const: str, title: str | None = None):
"""
:param reserved_const: Reserved constant, e.g. ``LAST_30_DAYS``.
:param title: Custom title, defaults to ``spec`` when not set.
"""
self._reserved_const = reserved_const
self._title = title if title is not None else reserved_const
def __repr__(self):
return f'During(reserved_const=({self._reserved_const}, title=({self._title}))'
[docs]
def get_title(self):
return self._title
[docs]
def to_gaql(self) -> str:
"""
Transforms to GAQL query string.
:return: GAQL fragment.
"""
return f'segments.date DURING {self._reserved_const}'
def __eq__(self, other):
if isinstance(other, During):
return self._reserved_const == other._reserved_const
return False
During.TODAY = During('TODAY', 'Today')
During.YESTERDAY = During('YESTERDAY', 'Yesterday')
During.LAST_7_DAYS = During('LAST_7_DAYS', 'Last 7 Days')
During.THIS_WEEK_SUN_TODAY = During('THIS_WEEK_SUN_TODAY', 'This Week (starting Sunday)')
During.THIS_WEEK_MON_TODAY = During('THIS_WEEK_MON_TODAY', 'This Week (starting Monday)')
During.LAST_14_DAYS = During('LAST_14_DAYS', 'Last 14 Days')
During.LAST_30_DAYS = During('LAST_30_DAYS', 'Last 30 Days')
During.LAST_BUSINESS_WEEK = During('LAST_BUSINESS_WEEK', 'Last Business Week')
During.LAST_WEEK_SUN_SAT = During('LAST_WEEK_SUN_SAT', 'Last Week (starting Sunday)')
During.LAST_WEEK_MON_SUN = During('LAST_WEEK_MON_SUN', 'Last Week (starting Monday)')
During.THIS_MONTH = During('THIS_MONTH', 'This Month')
During.LAST_MONTH = During('LAST_MONTH', 'Last Month')