Skip to content

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 BasicFormatter

True
**kwargs str | bool | int

extra keyword arguments passed to set_with_dataframe

{}
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)