pytanis

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)

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

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

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)

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)

def create_ticket(self, ticket: NewTicket):
    return self.post("tickets", data=ticket.dict())

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()

def list_agents(self) -> List[Agent]:
    agents = self.get("agents")
    assert isinstance(agents, List)
    return [Agent.parse_obj(dct) for dct in agents]

def list_teams(self) -> List[Team]:
    teams = self.get("teams")
    assert isinstance(teams, List)
    return [Team.parse_obj(dct) for dct in teams]

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()

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)

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)

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)

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)

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

def me(self) -> Me:
    """Returns what Pretalx knows about myself"""
    result = self._get_one("/api/me")
    return Me.parse_obj(result)

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)

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)

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)

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)

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)

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)

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)

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)

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)

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)

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)

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)

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)

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)

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)