Google Postmaster Domains Datas

class googlepostmasterapi.logger.Logger(verbose: bool | None = False)

Base class to manage logger & conf

_logger

Protected. Logger

Type:: logging.Logger

_log_tpl

Protected. All part to create log template

Type:: dict

verbose

Verbose mode

Type:: bool

__init__(verbose: bool | None = False) → None

Default constructor

Parameters:: verbose (bool) – Optional. Verbose mode. Default : false

_create_log_tpl() → str

Create log template

Returns:: Log template
Return type:: str

_get_current_class_name() → str

Return current instanciated class name

Returns:: Class name
Return type:: str

_init_logger() → None: Init logger with all handlers

_init_logger_format() → Formatter

Init format template for log output

Returns:: Log template format
Return type:: logging.Formatter

_init_logger_stderr(log_format: Formatter) → StreamHandler

Create logger handler to : stderr

Parameters:: log_format (logging.Formatter) – Log template
Returns:: Logger handler on stderr
Return type:: logging.StreamHandler

_init_logger_stdout(log_format: Formatter) → StreamHandler

Create logger handler to : stdout

Parameters:: log_format (logging.Formatter) – Log template
Returns:: Logger handler on stdout
Return type:: logging.StreamHandler

_init_resources_logger() → None: Init resources used by system : init logger

_log_add_filter(logger: StreamHandler, level: int) → None

Add logger filter

Parameters:

logger (logging.StreamHandler) – Instance to add filter
level (int) – Max level for filter

_log_add_handler(logger: RootLogger | Logger, handler: StreamHandler) → None

Add logger handler

Parameters:

logger (logging.RootLogger|logging.Logger) – Instance to add handler
handler (logging.StreamHandler) – Handler

_log_get_logger(name: str | None = None) → RootLogger | Logger

Crete logger based on name

Parameters:: name (str) – Optional. Name to create logger. Default : None
Returns:: RootLogger if name is undefined. Logger otherwise
Return type:: logging.RootLogger|logging.Logger

_log_set_format(logger: StreamHandler, log_format: Formatter) → None

Set logger format

Parameters:

logger (logging.StreamHandler) – Instance to set format
log_format (logging.Formatter) – Log template

_log_set_level(logger: RootLogger | Logger | StreamHandler, level: int) → None

Set logger level

Parameters:

logger (logging.StreamHandler|logging.Logger|logging.RootLogger) – Instance to set level
level (int) – Level

write_error(logs: List[str]) → None

Write error information

Parameters:: logs (str<>) – List of logs to write

write_log(logs: List[str], force_verbose: bool | None = False) → bool

Write log information

Parameters:

logs (str<>) – List of logs to write
force_verbose (bool) – Optional. Force verbose mode. Overload verbose class properties. Default to False

Returns:

False if no verbose mode. True otherwise

Return type:

bool

class googlepostmasterapi.logger.LoggingFilter(level: int)

Filters all messages with level < self.max_level

max_level

Max log level value

Type:: int

__init__(level: int) → None

Default constructor

Parameters:: level (int) – Max log level value

filter(log: LogRecord) → bool

Filter a log by its level

Parameters:: log (logging.LogRecord) – Log to filter
Returns:: True if log level is lower than max level
Return type:: bool

class googlepostmasterapi.base.Base(verbose: bool | None = False)

Base class with common methods

__init__(verbose: bool | None = False) → None

Default constructor

Parameters:: verbose (bool) – Optional. Verbose mode. Default : False

extract_stat_value(value: dict | None = None) → Any

Extract the actual scalar/list value from a GPT v2 StatisticValue object

Parameters:: value (dict) – Optional. StatisticValue object from GPT v2 (doubleValue/floatValue/intValue/stringValue/stringList). Default : None
Returns:: Extracted value. None if nothing set
Return type:: mixed

googlepostmasterapi.base.MODULE_PATH = '/home/debian/modules/python-pub-gpostmaster-domains-data': Current module path

class googlepostmasterapi.gpt.GPostmaster(token: str, verbose: bool | None = False)

Download data from Google postmaster tools

_uri_tpl

Protected. Template to create a domain parent uri

Type:: str

_compliance_uri_tpl

Protected. Template to create a domain compliance status uri

Type:: str

_delivery_error_reasons

Protected. Assoc of error_type to the list of error_reason supported by GPT v2

Type:: dict

_domains

Protected. All domains fetch from GPT

Type:: str[]

_service

Protected. Connector to GPT

Type:: googleapiclient.discovery.build

_parser

Protected. Connector to data cleaner

Type:: FlatData

_stats

Protected. Connector to statistiques data

Type:: Stats

scopes

GPT scopes to read data

Type:: str[]

__init__(token: str, verbose: bool | None = False) → None

Default constructor

Parameters:

token (str) – Local file path to GPT json token
verbose (bool) – Optional. Verbose mode. Default : False

_clean_domain_infos(key: str, data: dict) → dict

Clean domain infos

Parameters:

key (str) – Key to identify data on cleaner
data (dict) – Domain infos to clean

Returns:

Cleaned data

Return type:

dict

_clean_pool_returns(data: List[dict]) → List[dict]

Clean result from pool map returns : remove all state==false

Parameters:: data (dict[]) – List of dict from pool map
Returns:: List of dict from pool map with only state==true
Return type:: dict[]

_create_compliance_uri(domain: str) → str

Create uri to a domain compliance status to query

Parameters:: domain (str) – Domain to query
Returns:: Compliance status uri, format : domains/{domain}/complianceStatus
Return type:: str

_create_domain_uri(domain: str) → str

Create parent uri to a domain to query

Parameters:: domain (str) – Domain to query
Returns:: Parent uri, format : domains/{domain}
Return type:: str

_create_fbl_metric_definitions(fbl_ids: List[str]) → List[dict]

Create the list of metric definitions to query the spam rate of each feedback loop id

Parameters:: fbl_ids (str[]) – Feedback loop ids to query
Returns:: List of MetricDefinition objects
Return type:: dict[]

_create_metric_definitions() → List[dict]

Create the list of metric definitions to query on domainStats : spam rate / auth / tls inbound / delivery errors / feedback loop id

Returns:: List of MetricDefinition objects
Return type:: dict[]

_create_pool_data(input_date: str) → List[dict]

Create data to map call on pool with all domains

Parameters:: input_date (str) – Date to query
Returns:: List of dict with domain&input_date
Return type:: dict[]

_create_query_request(input_date: str, metric_definitions: List[dict]) → dict

Create a QueryDomainStatsRequest body

Parameters:

input_date (str) – Date to query, format : YYYYMMDD
metric_definitions (dict[]) – List of MetricDefinition objects to query

Returns:

QueryDomainStatsRequest body

Return type:

dict

_extract_fbl_ids(domain_stats: List[dict]) → List[str]

Extract the unique feedback loop ids found in a domainStats query result

Parameters:: domain_stats (dict[]) – List of DomainStat objects
Returns:: Unique feedback loop ids, in order of appearance
Return type:: string[]

_get_domain_infos_pool(data: dict) → dict

Abstract call to get_domain_infos from pool with data as dict

Parameters:: data (dict) – Values domain/input_date to send to get_domain_infos
Returns:: Result from get_domain_infos calls
Return type:: dict

_gpt_create_domain(domain: str) → bool

Call GPT to create a domain

Parameters:: domain (str) – Domain to create
Returns:: False if an error occurs during creation. True otherwise
Return type:: bool

_gpt_get_compliance_status(domain: str) → dict

Call GPT to get the compliance status of a domain

Parameters:: domain (str) – Domain to query
Returns:: DomainComplianceStatus object
Return type:: dict

_gpt_get_domain_info(domain: str, input_date: str) → dict

Call GPT to get all infos to a domain

Parameters:

domain (str) – Domain to query
input_date (str) – Date to query, format : YYYYMMDD

Returns:

Process state & result

Return type:

dict

_gpt_get_domain_verify_token(domain: str) → str

Call GPT to get a verification token to a domain. Be careful, GPT give a verification token even if the domain is not created

Parameters:: domain (str) – Domain to get verification domain
Returns:: Token to current domain
Return type:: str

_gpt_get_domains(next_page: str | None = None) → List[dict]

Call GPT to get all domains. Recursive call on pagination

Parameters:: next_page (str) – Optional. Token to get next page of domains. Default : None
Returns:: List of dict with all domains, format : [ { ‘name’: …, ‘createTime’: …, ‘permission’: … } ]
Return type:: list

_gpt_verify_domain(domain: str) → bool

Call GPT to verify a domain

Parameters:: domain (str) – Domain to check
Returns:: False if the domain is not verified. True otherwise
Return type:: bool

_init_parser_con() → None: Init data parser/cleaner con

_init_resources(token: str) → None

Init resources used by system : init service / parser / stats

Parameters:: token (str) – Local file path to GPT json token

_init_service(token: str) → None

Init service connector

Parameters:: token (str) – Local file path to GPT json token

_init_stats_con() → None: Init stats con. Should be manager by multiprocessing to work with pool

_load_token(token: str) → Credentials

Load GPT token

Parameters:: token (str) – Local file path to GPT json token
Returns:: GPT creds
Return type:: Credentials

_parse_input_date(input_date: str) → dict

Parse an input date to a GPT Date object

Parameters:: input_date (str) – Date to query, format : YYYYMMDD
Returns:: GPT Date object, format : { ‘year’: int, ‘month’: int, ‘day’: int }
Return type:: dict

_print_stats() → None: Display calls statistics

_query_domain_stats(parent: str, body: dict, page_token: str | None = None) → List[dict]

Call GPT domainStats.query. Recursive call on pagination

Parameters:

parent (str) – Parent uri of the domain to query
body (dict) – QueryDomainStatsRequest body
page_token (str) – Optional. Token to get next page of stats. Default : None

Returns:

List of DomainStat objects

Return type:

dict[]

_recursive_call(method: str, *args: Any, **kargs: Any) → Any

Method to abstract recursive call

Parameters:

method (string) – Mehtod name to call on current instance
args (mixed()) – Arguments to send to method
kargs (dict) – Keyword arguments to send to method

Returns:

Method returns

Return type:

mixed

create_domain(domain: str) → dict

Create a domain, then get the verification token

Parameters:: domain (str) – Domain to create & get verification domain
Returns:: state & token
Return type:: dict

get_all_domains_infos(input_date: str, pool_size: int | None = 2) → List[dict]

Call GPT on all available domains

Parameters:

input_date (str) – Date to query, format : YYYYMMDD
pool_size (int) – Optional. Number of threads. Default : 2

Returns:

All domain infos

Return type:

list

get_domain_infos(domain: str, input_date: str, print_stats: bool | None = True) → dict

Get infos to a domain

Parameters:

domain (str) – Domain to query
input_date (str) – Date to query, format : YYYYMMDD
print_stats (bool) – Optional. True to display stats of the call. Defaut : True

Returns:

Process state & domain infos

Return type:

dict

get_domain_verify_token(domain: str) → str

Get a verification token for a domainBe careful, GPT give a verification token even if the domain is not created

Parameters:: domain (str) – Domain to get verification domain
Returns:: Token to current domain
Return type:: str

get_domains() → None: Get all domains with permissions : owner/reader

verify_domain(domain: str) → bool

Verify a domain

Parameters:: domain (str) – Domain to check
Returns:: False if the domain is not verified. True otherwise
Return type:: bool

class googlepostmasterapi.data.FlatData

Clean data traffic stats from Google Postmaster Tools

data

Data cleaned

Type:: dict

_data_tpl

Protected. Template to clean data

Type:: dict

_compliance_status_tpl

Protected. Assoc to translate GPT compliance status to a lowercase string

Type:: dict

__init__() → None: Default constructor

_index_domain_stats(domain_stats: List[dict]) → dict

Index a list of DomainStat objects by their metric name

Parameters:: domain_stats (dict[]) – List of DomainStat objects
Returns:: Assoc metric name => extracted value
Return type:: dict

_parse_crypted_inbound(key: str, value: float | None = None) → bool

Clean part of metric : tls_inbound

Parameters:

key (str) – Key to identify data
value (float) – Optional. Inbound TLS encryption rate to convert to percent. Default : None

Returns:

False if no value to convert. True otherwise

Return type:

bool

_parse_delivery_err(key: str, value: dict) → bool

Clean part of metrics : delivery_error__{class}__{type}

Parameters:

key (str) – Key to identify data
value (dict) – Assoc ‘{class}__{type}’ => error rate

Returns:

False if no value to convert. True otherwise

Return type:

bool

_parse_domain_compliance(key: str, value: dict | None = None) → bool

Clean part of key : complianceStatus.complianceData

Parameters:

key (str) – Key to identify data
value (dict) – Optional. DomainComplianceData object. Default : None

Returns:

False if no value to convert. True otherwise

Return type:

bool

_parse_fbl(key: str, value: dict) → bool

Clean part of metric : feedback_loop_spam_rate__{id}

Parameters:

key (str) – Key to identify data
value (dict) – Assoc feedback loop id => spam rate

Returns:

False if no value to convert. True otherwise

Return type:

bool

_parse_status(status: dict | None = None) → str | None

Clean a ComplianceStatus object to a lowercase string

Parameters:: status (dict) – Optional. ComplianceStatus object. Default : None
Returns:: Cleaned status. None if no status set
Return type:: str

_parse_use_auth(key: str, **kargs: dict) → bool

Clean part of metrics : auth_dkim / auth_spf / auth_dmarc

Parameters:

key (str) – Key to identify data
dkim (float) – Optional. DKIM success rate to convert to percent
spf (float) – Optional. SPF success rate to convert to percent
dmarc (float) – Optional. DMARC success rate to convert to percent

Returns:

True if at leat one the three key exists. False otherwise

Return type:

bool

_parse_user_report_spam(key: str, value: float | None = None) → bool

Clean part of metric : spam_rate

Parameters:

key (str) – Key to identify data
value (float) – Optional. Spam rate to convert to percent. Default : None

Returns:

False if no value to convert. True otherwise

Return type:

bool

parse(key: str, data: dict) → dict

Parse data from GPT to a flatern version with all values

Parameters:

key (str) – Key to identify data
data (dict) – Data from GPT to clean, format : { ‘domainStats’: [ … ], ‘complianceStatus’: { … } }

Returns:

Cleaned data from GPT

Return type:

dict

class googlepostmasterapi.stats.Stats

Process stats of GPostmaster

data

Statistics from domains infos download

Type:: dict

__init__() → None: Default constructor

add_err() → None: Increment counters : total + err

add_err_http(code: int, err: str, domain: str) → None

Increment counters : total + err & add domain to http error data

Parameters:

code (int) – Http code
err (string) – Http error message
domain (string) – Domain to add to http error

add_ok() → None: Increment counters : total + ok

add_total() → None: Increment counter : total

print_stats() → bool

Display calls statistics

Returns:: False if nothing to display. True otherwise
Return type:: bool

Google Postmaster Domains Datas

Indices and tables