Reference
pytanis
¶
__all__ = ['GSheetsClient', 'HelpDeskClient', 'PretalxClient', '__version__', 'get_cfg']
module-attribute
¶
__version__ = version('pytanis')
module-attribute
¶
GSheetsClient(config: Config | None = 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: Config | None = 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)
wrange = worksheet_range(worksheet)
try:
worksheet.clear()
worksheet.clear_basic_filter()
format_cell_range(worksheet, wrange, default_fmt)
rules = get_conditional_format_rules(worksheet)
rules.clear()
rules.save()
set_data_validation_for_cell_range(worksheet, wrange, None)
except APIError as error:
self._exception_feedback(error)
gsheet(spreadsheet_id: str, worksheet_name: str | None = None, *, create_ws: bool = False) -> 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: str | None = None, *, create_ws: bool = False
) -> 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
elif 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: 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: 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: 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 | 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 | 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: 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)
params = {'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: Config | None = None)
¶
Source code in src/pytanis/helpdesk/client.py
def __init__(self, config: Config | None = 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(calls=1, seconds=10) # Helpdesk is really strange when it comes to this
create_ticket(ticket: NewTicket)
¶
Source code in src/pytanis/helpdesk/client.py
def create_ticket(self, ticket: NewTicket):
return self.post('tickets', data=ticket.model_dump())
get(endpoint: str, params: QueryParams | None = 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: QueryParams | None = 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')
if not isinstance(agents, list):
msg = 'Received JSON is not a list object'
raise ValueError(msg)
return [Agent.model_validate(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')
if not isinstance(teams, list):
msg = 'Received JSON is not a list object'
raise ValueError(msg)
return [Team.model_validate(dct) for dct in teams]
post(endpoint: str, data: dict[str, Any], params: QueryParams | None = None) -> JSON
¶
Source code in src/pytanis/helpdesk/client.py
def post(self, endpoint: str, data: dict[str, Any], params: QueryParams | None = 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: Config | None = None, *, blocking: bool = False)
¶
Client for the Pretalx API
Source code in src/pytanis/pretalx/client.py
def __init__(self, config: Config | None = 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(calls=2, seconds=1) # we are nice by default and Pretalx doesn't allow many calls at once.
blocking = blocking
instance-attribute
¶
answer(event_slug: str, id: int, *, params: QueryParams | None = None) -> Answer
¶
Returns a specific answer
Source code in src/pytanis/pretalx/client.py
def answer(self, event_slug: str, id: int, *, params: QueryParams | None = None) -> Answer: # noqa: A002
"""Returns a specific answer"""
return self._endpoint_id(Answer, event_slug, 'answers', id, params=params)
answers(event_slug: str, *, params: QueryParams | None = 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: QueryParams | None = 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: QueryParams | None = None) -> Event
¶
Returns detailed information about a specific event
Source code in src/pytanis/pretalx/client.py
def event(self, event_slug: str, *, params: QueryParams | None = 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.model_validate(result)
events(*, params: QueryParams | None = None) -> tuple[int, Iterator[Event]]
¶
Lists all events and their details
Source code in src/pytanis/pretalx/client.py
def events(self, *, params: QueryParams | None = 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.model_validate(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.model_validate(result)
question(event_slug: str, id: int, *, params: QueryParams | None = None) -> Question
¶
Returns a specific question
Source code in src/pytanis/pretalx/client.py
def question(self, event_slug: str, id: int, *, params: QueryParams | None = None) -> Question: # noqa: A002
"""Returns a specific question"""
return self._endpoint_id(Question, event_slug, 'questions', id, params=params)
questions(event_slug: str, *, params: QueryParams | None = 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: QueryParams | None = 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: QueryParams | None = None) -> Review
¶
Returns a specific review
Source code in src/pytanis/pretalx/client.py
def review(self, event_slug: str, id: int, *, params: QueryParams | None = None) -> Review: # noqa: A002
"""Returns a specific review"""
return self._endpoint_id(Review, event_slug, 'reviews', id, params=params)
reviews(event_slug: str, *, params: QueryParams | None = 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: QueryParams | None = 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: QueryParams | None = None) -> Room
¶
Returns a specific room
Source code in src/pytanis/pretalx/client.py
def room(self, event_slug: str, id: int, *, params: QueryParams | None = None) -> Room: # noqa: A002
"""Returns a specific room"""
return self._endpoint_id(Room, event_slug, 'rooms', id, params=params)
rooms(event_slug: str, *, params: QueryParams | None = 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: QueryParams | None = 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: QueryParams | None = None) -> Speaker
¶
Returns a specific speaker
Source code in src/pytanis/pretalx/client.py
def speaker(self, event_slug: str, code: str, *, params: QueryParams | None = None) -> Speaker:
"""Returns a specific speaker"""
return self._endpoint_id(Speaker, event_slug, 'speakers', code, params=params)
speakers(event_slug: str, *, params: QueryParams | None = 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: QueryParams | None = 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: QueryParams | None = None) -> Submission
¶
Returns a specific submission
Source code in src/pytanis/pretalx/client.py
def submission(self, event_slug: str, code: str, *, params: QueryParams | None = None) -> Submission:
"""Returns a specific submission"""
return self._endpoint_id(Submission, event_slug, 'submissions', code, params=params)
submissions(event_slug: str, *, params: QueryParams | None = 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: QueryParams | None = 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: QueryParams | None = None) -> Tag
¶
Returns a specific tag
Source code in src/pytanis/pretalx/client.py
def tag(self, event_slug: str, tag: str, *, params: QueryParams | None = None) -> Tag:
"""Returns a specific tag"""
return self._endpoint_id(Tag, event_slug, 'tags', tag, params=params)
tags(event_slug: str, *, params: QueryParams | None = 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: QueryParams | None = 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: QueryParams | None = None) -> Talk
¶
Returns a specific talk
Source code in src/pytanis/pretalx/client.py
def talk(self, event_slug: str, code: str, *, params: QueryParams | None = None) -> Talk:
"""Returns a specific talk"""
return self._endpoint_id(Talk, event_slug, 'talks', code, params=params)
talks(event_slug: str, *, params: QueryParams | None = 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: QueryParams | None = 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.model_validate(cfg_dict)