style: type hint
This commit is contained in:
parent
9ed2beacec
commit
65f4aa9e4c
39
bin/utils.py
39
bin/utils.py
|
@ -4,6 +4,8 @@
|
|||
Common functions used in this add-on
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from configparser import ConfigParser
|
||||
from csv import QUOTE_ALL, DictReader
|
||||
from os import environ, path
|
||||
|
@ -15,7 +17,7 @@ import requests
|
|||
class Utility:
|
||||
"""Provide common functions"""
|
||||
|
||||
def __get_proxy(self, url):
|
||||
def __get_proxy(self, url: str) -> dict[str, dict[str, str]] | str:
|
||||
"""
|
||||
Determine http proxy setting of a URL according to Splunk server configuration.
|
||||
Return {dict} of http/https proxy value if a URL should be proxied.
|
||||
|
@ -57,13 +59,12 @@ class Utility:
|
|||
|
||||
return {}
|
||||
|
||||
def download(self, urls, index=0):
|
||||
def download(self, urls: list | tuple | str, index: int = 0) -> str:
|
||||
"""
|
||||
Send a GET request to the URL and return content of the response.
|
||||
|
||||
Arguments:
|
||||
urls {list/tuple/string} -- A list of URLs to try in sequence
|
||||
index -- List's index to start
|
||||
:param urls: A list of URLs to try in sequence
|
||||
:param index: List's index to start
|
||||
"""
|
||||
if isinstance(urls, str):
|
||||
urls = (urls,)
|
||||
|
@ -88,7 +89,7 @@ class Utility:
|
|||
except requests.exceptions.RequestException as err:
|
||||
raise err
|
||||
|
||||
def __split_column(self, input_str=None):
|
||||
def __split_column(self, input_str: str | list | None = None) -> list[str] | list:
|
||||
"""Split {string} into {list} using comma separator"""
|
||||
if isinstance(input_str, str):
|
||||
return [x.strip() for x in input_str.split(",")]
|
||||
|
@ -96,24 +97,30 @@ class Utility:
|
|||
return input_str
|
||||
return []
|
||||
|
||||
def insert_affix(self, row, prefix_opt=None, suffix_opt=None, affix_opt=None):
|
||||
def insert_affix(
|
||||
self,
|
||||
row: dict,
|
||||
prefix_opt: str | list | None = None,
|
||||
suffix_opt: str | list | None = None,
|
||||
affix_opt: str | list | None = None,
|
||||
) -> dict:
|
||||
"""
|
||||
Affix wildcard "*" character to existing values
|
||||
|
||||
Arguments:
|
||||
row {dict} -- A row of an array-parsed CSV
|
||||
prefix_opt {string/list} -- A column name or a comma-separated list of column names to have
|
||||
wildcard prefixed to their non-empty value.
|
||||
suffix_opt {string/list} -- Same as prefix_opt but have the wildcard suffixed instead.
|
||||
affix_opt {string/list} -- Same as prefix_opt but have the wildcard prefixed and suffixed.
|
||||
:param row: A row of an array-parsed CSV
|
||||
:param prefix_opt: A column name or a comma-separated list of column names to have
|
||||
wildcard prefixed to their non-empty value.
|
||||
:param suffix_opt: Same as prefix_opt but have the wildcard suffixed instead.
|
||||
:param affix_opt: Same as prefix_opt but have the wildcard prefixed and suffixed.
|
||||
|
||||
Return:
|
||||
A new row with prefix/suffix columns appended
|
||||
Return a new row with prefix/suffix columns appended
|
||||
"""
|
||||
|
||||
prefix_opt_list = self.__split_column(prefix_opt)
|
||||
suffix_opt_list = self.__split_column(suffix_opt)
|
||||
affix_opt_list = self.__split_column(affix_opt)
|
||||
new_column = {}
|
||||
|
||||
for column in prefix_opt_list:
|
||||
if column in row and len(row[column]) >= 1:
|
||||
new_column = {
|
||||
|
@ -135,7 +142,7 @@ class Utility:
|
|||
|
||||
return {**row, **new_column}
|
||||
|
||||
def csv_reader(self, csv_str):
|
||||
def csv_reader(self, csv_str: str) -> DictReader:
|
||||
"""Parse an CSV input string into an interable of {dict} rows whose keys correspond to column names"""
|
||||
return DictReader(
|
||||
filter(lambda row: row[0] != "#", csv_str.splitlines()), quoting=QUOTE_ALL
|
||||
|
|
4
build.py
4
build.py
|
@ -12,7 +12,7 @@ from subprocess import check_call
|
|||
from sys import executable
|
||||
|
||||
|
||||
def version():
|
||||
def version() -> str:
|
||||
"""
|
||||
Return version number from app.conf or commit hash if in CI
|
||||
"""
|
||||
|
@ -46,7 +46,7 @@ def version():
|
|||
return launcher.get("version", "")
|
||||
|
||||
|
||||
def exclusion(tarinfo):
|
||||
def exclusion(tarinfo: tarfile.TarInfo) -> tarfile.TarInfo | None:
|
||||
"""Exclude dev files and cache, and reset file stats"""
|
||||
|
||||
# exclude certain folders/files
|
||||
|
|
Loading…
Reference in New Issue