Reference
pytanis
¶
__all__ = ['__version__', 'GSheetClient', 'PretalxClient', 'HelpDeskClient', 'get_cfg']
module-attribute
¶
__version__ = version('pytanis')
module-attribute
¶
GSheetClient(config: Optional[Config] = None, read_only: bool = True)
¶
Google API to easily handle GSheets and other files on GDrive
By default, only the least permissive scope GSHEET_RO
in case of read_only = True
is used.
Source code in src/pytanis/google.py
def __init__(self, config: Optional[Config] = None, read_only: bool = True):
self._read_only = read_only
if read_only:
self._scopes = [Scope.GSHEET_RO]
else:
self._scopes = [Scope.GSHEET_RW]
if config is None:
config = get_cfg()
self._config = config
self.gc = gspread_client(self._scopes, config) # gspread client for more functionality
gc = gspread_client(self._scopes, config)
instance-attribute
¶
clear_gsheet(spreadsheet_id: str, worksheet_name: str)
¶
Clear the worksheet including values, formatting, filtering, etc.
Source code in src/pytanis/google.py
def clear_gsheet(self, spreadsheet_id: str, worksheet_name: str):
"""Clear the worksheet including values, formatting, filtering, etc."""
worksheet = self.gsheet(spreadsheet_id, worksheet_name, create_ws=False)
default_fmt = get_default_format(worksheet.spreadsheet)
range = worksheet_range(worksheet)
try:
worksheet.clear()
worksheet.clear_basic_filter()
format_cell_range(worksheet, range, default_fmt)
rules = get_conditional_format_rules(worksheet)
rules.clear()
rules.save()
set_data_validation_for_cell_range(worksheet, range, None)
except APIError as error:
self._exception_feedback(error)
gsheet(spreadsheet_id: str, worksheet_name: Optional[str] = None, create_ws: bool = False) -> Union[Worksheet, Spreadsheet]
¶
Retrieve a Google sheet by its id and the name
Open a Google sheet in your browser and check the URL to retrieve the id, e.g.: https://docs.google.com/spreadsheets/d/SPREEDSHEET_ID/edit...
If the spreadsheet as several worksheets (check the lower bar) then worksheet_name
can be used to specify a specific one.
Source code in src/pytanis/google.py
def gsheet(
self, spreadsheet_id: str, worksheet_name: Optional[str] = None, create_ws: bool = False
) -> Union[Worksheet, Spreadsheet]:
"""Retrieve a Google sheet by its id and the name
Open a Google sheet in your browser and check the URL to retrieve the id, e.g.:
https://docs.google.com/spreadsheets/d/SPREEDSHEET_ID/edit...
If the spreadsheet as several worksheets (check the lower bar) then `worksheet_name` can be used to
specify a specific one.
"""
spreadsheet = self.gc.open_by_key(spreadsheet_id)
if worksheet_name is None:
return spreadsheet
else:
if worksheet_name in [ws.title for ws in spreadsheet.worksheets()]:
return spreadsheet.worksheet(worksheet_name)
elif create_ws:
worksheet = spreadsheet.add_worksheet(title=worksheet_name, rows=100, cols=20)
self._wait_for_worksheet(spreadsheet_id, worksheet_name)
return worksheet
else:
return spreadsheet.worksheet(worksheet_name) # raises exception
gsheet_as_df(spreadsheet_id: str, worksheet_name: str, kwargs: Union[str, bool, int]) -> pd.DataFrame
¶
Returns a worksheet as dataframe
Source code in src/pytanis/google.py
def gsheet_as_df(self, spreadsheet_id: str, worksheet_name: str, **kwargs: Union[str, bool, int]) -> pd.DataFrame:
"""Returns a worksheet as dataframe"""
worksheet = self.gsheet(spreadsheet_id, worksheet_name)
df = get_as_dataframe(worksheet, **kwargs)
# remove Nan rows & columns as they are exported by default
df.dropna(how='all', inplace=True, axis=0)
df.dropna(how='all', inplace=True, axis=1)
return df
recreate_token()
¶
Recreate the current token using the scopes given at initialization
Source code in src/pytanis/google.py
def recreate_token(self):
"""Recreate the current token using the scopes given at initialization"""
self._config.Google.token_json.unlink(missing_ok=True)
self.gc = gspread_client(self._scopes, self._config)
save_df_as_gsheet(df: pd.DataFrame, spreadsheet_id: str, worksheet_name: str, create_ws: bool = False, default_fmt: bool = True, kwargs: Union[str, bool, int])
¶
Save the given dataframe as worksheet in a spreadsheet
Make sure that the scope passed gives you write permissions
Parameters:
Name | Type | Description | Default |
---|---|---|---|
df | pd.DataFrame | dataframe to save | required |
spreadsheet_id | str | id of the Google spreadsheet | required |
worksheet_name | str | name of the worksheet within the spreadsheet | required |
create_ws | bool | create the worksheet if non-existent | False |
default_fmt | bool | apply default formatter | True |
**kwargs | Union[str, bool, int] | extra keyword arguments passed to | {} |
Source code in src/pytanis/google.py
def save_df_as_gsheet(
self,
df: pd.DataFrame,
spreadsheet_id: str,
worksheet_name: str,
create_ws: bool = False,
default_fmt: bool = True,
**kwargs: Union[str, bool, int],
):
"""Save the given dataframe as worksheet in a spreadsheet
Make sure that the scope passed gives you write permissions
Args:
df: dataframe to save
spreadsheet_id: id of the Google spreadsheet
worksheet_name: name of the worksheet within the spreadsheet
create_ws: create the worksheet if non-existent
default_fmt: apply default formatter `BasicFormatter`
**kwargs: extra keyword arguments passed to `set_with_dataframe`
"""
worksheet = self.gsheet(spreadsheet_id, worksheet_name, create_ws=create_ws)
# make sure it's really only the dataframe, not some residue
self.clear_gsheet(spreadsheet_id, worksheet_name)
# ToDo: Starting from Python 3.9 on just use the | operator
params = {**dict(resize=True), **dict(**kwargs)} # set sane defaults
try:
set_with_dataframe(worksheet, df, **params)
if default_fmt:
format_with_dataframe(worksheet, df)
except APIError as error:
self._exception_feedback(error)
HelpDeskClient(config: Optional[Config] = None)
¶
Source code in src/pytanis/helpdesk/client.py
def __init__(self, config: Optional[Config] = None):
if config is None:
config = get_cfg()
self._config = config
# Important: Always use a custom User-Agent, never a generic one.
# Generic User-Agents are filtered by helpdesk to reduce spam.
self._headers = {"User-Agent": "Pytanis"}
self._get_throttled = self._get
self._post_throttled = self._post
self.set_throttling(2, 1) # we are nice by default
create_ticket(ticket: NewTicket)
¶
Source code in src/pytanis/helpdesk/client.py
def create_ticket(self, ticket: NewTicket):
return self.post("tickets", data=ticket.dict())
get(endpoint: str, params: Optional[Dict[str, str]] = None) -> JSON
¶
Retrieve data via throttled GET request and return the JSON
Source code in src/pytanis/helpdesk/client.py
def get(self, endpoint: str, params: Optional[Dict[str, str]] = None) -> JSON:
"""Retrieve data via throttled GET request and return the JSON"""
resp = self._get_throttled(endpoint, params)
resp.raise_for_status()
return resp.json()
list_agents() -> List[Agent]
¶
Source code in src/pytanis/helpdesk/client.py
def list_agents(self) -> List[Agent]:
agents = self.get("agents")
assert isinstance(agents, List)
return [Agent.parse_obj(dct) for dct in agents]
list_teams() -> List[Team]
¶
Source code in src/pytanis/helpdesk/client.py
def list_teams(self) -> List[Team]:
teams = self.get("teams")
assert isinstance(teams, List)
return [Team.parse_obj(dct) for dct in teams]
post(endpoint: str, data: Dict[str, Any], params: Optional[Dict[str, str]] = None) -> JSON
¶
Source code in src/pytanis/helpdesk/client.py
def post(self, endpoint: str, data: Dict[str, Any], params: Optional[Dict[str, str]] = None) -> JSON:
resp = self._post_throttled(endpoint, data, params)
resp.raise_for_status()
return resp.json()
set_throttling(calls: int, seconds: int)
¶
Throttle the number of calls per seconds to the Pretalx API
Source code in src/pytanis/helpdesk/client.py
def set_throttling(self, calls: int, seconds: int):
"""Throttle the number of calls per seconds to the Pretalx API"""
_logger.debug("throttling", calls=calls, seconds=seconds)
self._get_throttled = throttle(calls, seconds)(self._get)
self._post_throttled = throttle(calls, seconds)(self._post)
PretalxClient(config: Optional[Config] = None, blocking: bool = False)
¶
Client for the Pretalx API
Source code in src/pytanis/pretalx/client.py
def __init__(self, config: Optional[Config] = None, blocking: bool = False):
if config is None:
config = get_cfg()
self._config = config
self._get_throttled = self._get
self.blocking = blocking
self.set_throttling(1, 2) # we are nice by default
blocking = blocking
instance-attribute
¶
answer(event_slug: str, id: int, *, params: Optional[QueryParamType] = None) -> Answer
¶
Returns a specific answer
Source code in src/pytanis/pretalx/client.py
def answer(self, event_slug: str, id: int, *, params: Optional[QueryParamType] = None) -> Answer:
"""Returns a specific answer"""
return self._endpoint_id(Answer, event_slug, "answers", id, params=params)
answers(event_slug: str, *, params: Optional[QueryParamType] = None) -> Tuple[int, Iterator[Answer]]
¶
Lists all answers and their details
Source code in src/pytanis/pretalx/client.py
def answers(self, event_slug: str, *, params: Optional[QueryParamType] = None) -> Tuple[int, Iterator[Answer]]:
"""Lists all answers and their details"""
return self._endpoint_lst(Answer, event_slug, "answers", params=params)
event(event_slug: str, *, params: Optional[QueryParamType] = None) -> Event
¶
Returns detailed information about a specific event
Source code in src/pytanis/pretalx/client.py
def event(self, event_slug: str, *, params: Optional[QueryParamType] = None) -> Event:
"""Returns detailed information about a specific event"""
endpoint = f"/api/events/{event_slug}/"
result = self._get_one(endpoint, params)
_logger.debug("result", resp=result)
return Event.parse_obj(result)
events(*, params: Optional[QueryParamType] = None) -> Tuple[int, Iterator[Event]]
¶
Lists all events and their details
Source code in src/pytanis/pretalx/client.py
def events(self, *, params: Optional[QueryParamType] = None) -> Tuple[int, Iterator[Event]]:
"""Lists all events and their details"""
count, results = self._get_many("/api/events/", params)
events = iter(_logger.debug("result", resp=r) or Event.parse_obj(r) for r in results)
return count, events
me() -> Me
¶
Returns what Pretalx knows about myself
Source code in src/pytanis/pretalx/client.py
def me(self) -> Me:
"""Returns what Pretalx knows about myself"""
result = self._get_one("/api/me")
return Me.parse_obj(result)
question(event_slug: str, id: int, *, params: Optional[QueryParamType] = None) -> Question
¶
Returns a specific question
Source code in src/pytanis/pretalx/client.py
def question(self, event_slug: str, id: int, *, params: Optional[QueryParamType] = None) -> Question:
"""Returns a specific question"""
return self._endpoint_id(Question, event_slug, "questions", id, params=params)
questions(event_slug: str, *, params: Optional[QueryParamType] = None) -> Tuple[int, Iterator[Question]]
¶
Lists all questions and their details
Source code in src/pytanis/pretalx/client.py
def questions(self, event_slug: str, *, params: Optional[QueryParamType] = None) -> Tuple[int, Iterator[Question]]:
"""Lists all questions and their details"""
return self._endpoint_lst(Question, event_slug, "questions", params=params)
review(event_slug: str, id: int, *, params: Optional[QueryParamType] = None) -> Review
¶
Returns a specific review
Source code in src/pytanis/pretalx/client.py
def review(self, event_slug: str, id: int, *, params: Optional[QueryParamType] = None) -> Review:
"""Returns a specific review"""
return self._endpoint_id(Review, event_slug, "reviews", id, params=params)
reviews(event_slug: str, *, params: Optional[QueryParamType] = None) -> Tuple[int, Iterator[Review]]
¶
Lists all reviews and their details
Source code in src/pytanis/pretalx/client.py
def reviews(self, event_slug: str, *, params: Optional[QueryParamType] = None) -> Tuple[int, Iterator[Review]]:
"""Lists all reviews and their details"""
return self._endpoint_lst(Review, event_slug, "reviews", params=params)
room(event_slug: str, id: int, *, params: Optional[QueryParamType] = None) -> Room
¶
Returns a specific room
Source code in src/pytanis/pretalx/client.py
def room(self, event_slug: str, id: int, *, params: Optional[QueryParamType] = None) -> Room:
"""Returns a specific room"""
return self._endpoint_id(Room, event_slug, "rooms", id, params=params)
rooms(event_slug: str, *, params: Optional[QueryParamType] = None) -> Tuple[int, Iterator[Room]]
¶
Lists all rooms and their details
Source code in src/pytanis/pretalx/client.py
def rooms(self, event_slug: str, *, params: Optional[QueryParamType] = None) -> Tuple[int, Iterator[Room]]:
"""Lists all rooms and their details"""
return self._endpoint_lst(Room, event_slug, "rooms", params=params)
set_throttling(calls: int, seconds: int)
¶
Throttle the number of calls per seconds to the Pretalx API
Source code in src/pytanis/pretalx/client.py
def set_throttling(self, calls: int, seconds: int):
"""Throttle the number of calls per seconds to the Pretalx API"""
_logger.info("throttling", calls=calls, seconds=seconds)
self._get_throttled = throttle(calls, seconds)(self._get)
speaker(event_slug: str, code: str, *, params: Optional[QueryParamType] = None) -> Speaker
¶
Returns a specific speaker
Source code in src/pytanis/pretalx/client.py
def speaker(self, event_slug: str, code: str, *, params: Optional[QueryParamType] = None) -> Speaker:
"""Returns a specific speaker"""
return self._endpoint_id(Speaker, event_slug, "speakers", code, params=params)
speakers(event_slug: str, *, params: Optional[QueryParamType] = None) -> Tuple[int, Iterator[Speaker]]
¶
Lists all speakers and their details
Source code in src/pytanis/pretalx/client.py
def speakers(self, event_slug: str, *, params: Optional[QueryParamType] = None) -> Tuple[int, Iterator[Speaker]]:
"""Lists all speakers and their details"""
return self._endpoint_lst(Speaker, event_slug, "speakers", params=params)
submission(event_slug: str, code: str, *, params: Optional[QueryParamType] = None) -> Submission
¶
Returns a specific submission
Source code in src/pytanis/pretalx/client.py
def submission(self, event_slug: str, code: str, *, params: Optional[QueryParamType] = None) -> Submission:
"""Returns a specific submission"""
return self._endpoint_id(Submission, event_slug, "submissions", code, params=params)
submissions(event_slug: str, *, params: Optional[QueryParamType] = None) -> Tuple[int, Iterator[Submission]]
¶
Lists all submissions and their details
Source code in src/pytanis/pretalx/client.py
def submissions(
self, event_slug: str, *, params: Optional[QueryParamType] = None
) -> Tuple[int, Iterator[Submission]]:
"""Lists all submissions and their details"""
return self._endpoint_lst(Submission, event_slug, "submissions", params=params)
tag(event_slug: str, tag: str, *, params: Optional[QueryParamType] = None) -> Tag
¶
Returns a specific tag
Source code in src/pytanis/pretalx/client.py
def tag(self, event_slug: str, tag: str, *, params: Optional[QueryParamType] = None) -> Tag:
"""Returns a specific tag"""
return self._endpoint_id(Tag, event_slug, "tags", tag, params=params)
tags(event_slug: str, *, params: Optional[QueryParamType] = None) -> Tuple[int, Iterator[Tag]]
¶
Lists all tags and their details
Source code in src/pytanis/pretalx/client.py
def tags(self, event_slug: str, *, params: Optional[QueryParamType] = None) -> Tuple[int, Iterator[Tag]]:
"""Lists all tags and their details"""
return self._endpoint_lst(Tag, event_slug, "tags", params=params)
talk(event_slug: str, code: str, *, params: Optional[QueryParamType] = None) -> Talk
¶
Returns a specific talk
Source code in src/pytanis/pretalx/client.py
def talk(self, event_slug: str, code: str, *, params: Optional[QueryParamType] = None) -> Talk:
"""Returns a specific talk"""
return self._endpoint_id(Talk, event_slug, "talks", code, params=params)
talks(event_slug: str, *, params: Optional[QueryParamType] = None) -> Tuple[int, Iterator[Talk]]
¶
Lists all talks and their details
Source code in src/pytanis/pretalx/client.py
def talks(self, event_slug: str, *, params: Optional[QueryParamType] = None) -> Tuple[int, Iterator[Talk]]:
"""Lists all talks and their details"""
return self._endpoint_lst(Talk, event_slug, "talks", params=params)
get_cfg() -> Config
¶
Returns the configuration as an object
Source code in src/pytanis/config.py
def get_cfg() -> Config:
"""Returns the configuration as an object"""
cfg_path = get_cfg_file()
with open(cfg_path, "rb") as fh:
cfg_dict = tomli.load(fh)
# add config path to later resolve relative paths of config values
cfg_dict["cfg_path"] = cfg_path
return Config.parse_obj(cfg_dict)