Source code for parsons.airmeet.airmeet

from parsons.etl.table import Table
from parsons.utilities import check_env
from parsons.utilities.api_connector import APIConnector

AIRMEET_DEFAULT_URI = "https://api-gateway.airmeet.com/prod/"


[docs] class Airmeet(object): """ Instantiate class. `Args:` airmeet_uri: string The URI of the Airmeet API endpoint. Not required. The default is https://api-gateway.airmeet.com/prod/. You can set an ``AIRMEET_URI`` env variable or use this parameter when instantiating the class. airmeet_access_key: string The Airmeet API access key. airmeet_secret_key: string The Airmeet API secret key. For instructions on how to generate an access key and secret key set, see `Airmeet's Event Details API documentation <https://help.airmeet.com/support/solutions/articles/82000909768-1-event-details-airmeet-public-api>`_. """ def __init__(self, airmeet_uri=None, airmeet_access_key=None, airmeet_secret_key=None): """ Authenticate with the Airmeet API and update the connection headers with the access token. `Args:` airmeet_uri: string The Airmeet API endpoint. airmeet_access_key: string The Airmeet API access key. airmeet_secret_key: string The Airmeet API secret key. `Returns:` ``None`` """ self.uri = check_env.check("AIRMEET_URI", airmeet_uri, optional=True) or AIRMEET_DEFAULT_URI self.client = APIConnector(self.uri) self.airmeet_client_key = check_env.check("AIRMEET_ACCESS_KEY", airmeet_access_key) self.airmeet_client_secret = check_env.check("AIRMEET_SECRET_KEY", airmeet_secret_key) self.client.headers = { "X-Airmeet-Access-Key": self.airmeet_client_key, "X-Airmeet-Secret-Key": self.airmeet_client_secret, } response = self.client.post_request(url="auth", success_codes=[200]) self.token = response["token"] # API calls expect the token in the header. self.client.headers = { "Content-Type": "application/json", "X-Airmeet-Access-Token": self.token, } def _get_all_pages(self, url, page_size=50, **kwargs) -> Table: """ Get all the results from an Airmeet API url, handling pagination based on the returned pageCount. `Args:` page_size: 50 The number of items to get per page. The max allowed varies by API call. For details, see `Airmeet's Event Details API documentation <https://help.airmeet.com/support/solutions/articles/82000909768-1-event-details-airmeet-public-api>`_. **kwargs: Additional parameters to include in the request. `Returns:` ``None`` """ results = [] cursor_after = "" # For getting the next set of results kwargs["size"] = page_size # Initial API call to get the first page of data response = self.client.get_request(url=url, params=kwargs) # Some APIs are asynchronous and will return a 202 if the request # should be tried again after five minutes, because the results # set needs to be built. if "statusCode" in response and response["statusCode"] != 200: raise Exception(response) else: results.extend(response["data"]) if "cursors" in response and response["cursors"]["pageCount"] > 1: cursor_after = response["cursors"]["after"] # Fetch subsequent pages if needed for _ in range(2, response["cursors"]["pageCount"] + 1): kwargs["after"] = cursor_after response = self.client.get_request(url=url, params=kwargs) results.extend(response["data"]) cursor_after = response["cursors"]["after"] return Table(results)
[docs] def list_airmeets(self) -> Table: """ Get the list of Airmeets. The API excludes any Airmeets that are Archived (Deleted). `Returns:` Parsons.Table List of Airmeets """ return self._get_all_pages(url="airmeets", page_size=500)
[docs] def fetch_airmeet_participants( self, airmeet_id, sorting_key="registrationDate", sorting_direction="DESC" ) -> Table: """ Get all participants (registrations) for a specific Airmeet, handling pagination based on the returned totalUserCount. This API doesn't use cursors for paging, so we can't use _get_all_pages() here. `Args:` airmeet_id: string The id of the Airmeet. sorting_key: string The key to sort the participants by. Can be 'name', 'email', or 'registrationDate' (the default). sorting_direction: string Can be either 'ASC' or 'DESC' (the default). `Returns:` Parsons.Table List of participants for the Airmeet event """ participants = [] # List to hold all participants page_size = 1000 # Maximum number of results per page # Initial API call to get the total user count and first page of data response = self.client.get_request( url=f"airmeet/{airmeet_id}/participants", params={ "pageNumber": 1, "resultSize": page_size, "sortingKey": sorting_key, "sortingDirection": sorting_direction, }, ) participants.extend(response["participants"]) # Calculate total pages needed based on totalUserCount. total_count = response["totalUserCount"] total_pages = (total_count + page_size - 1) // page_size # This rounds up the division. # Fetch subsequent pages if needed. for page in range(2, total_pages + 1): response = self.client.get_request( url=f"airmeet/{airmeet_id}/participants", params={ "pageNumber": page, "resultSize": page_size, "sortingKey": sorting_key, "sortingDirection": sorting_direction, }, ) participants.extend(response["participants"]) return Table(participants)
[docs] def fetch_airmeet_sessions(self, airmeet_id) -> Table: """ Get the list of sessions for an Airmeet. `Args:` airmeet_id: string The id of the Airmeet. `Returns:` Parsons.Table List of sessions for this Airmeet event """ response = self.client.get_request(url=f"airmeet/{airmeet_id}/info") return Table(response["sessions"])
[docs] def fetch_airmeet_info(self, airmeet_id, lists_to_tables=False): """ Get the data for an Airmeet (event), which include the list of sessions, session hosts/cohosts, and various other info. `Args:` airmeet_id: string The id of the Airmeet. lists_to_tables: bool If True, will convert any dictionary values that are lists to Tables. `Returns:` Dict containing the Airmeet data """ response = self.client.get_request(url=f"airmeet/{airmeet_id}/info") if lists_to_tables: for k in response: if isinstance(response[k], list): response[k] = Table(response[k]) return response
[docs] def fetch_airmeet_custom_registration_fields(self, airmeet_id) -> Table: """ Get the list of custom registration fields for an Airmeet. `Args:` airmeet_id: string The id of the Airmeet. `Returns:` Parsons.Table List of custom registration fields for this Airmeet event """ response = self.client.get_request(url=f"airmeet/{airmeet_id}/custom-fields") return Table(response["customFields"])
[docs] def fetch_event_attendance(self, airmeet_id) -> Table: """ Get all attendees for an Airmeet, handling pagination based on the returned pageCount. Results include attendance only from sessions with a status of `FINISHED`. Maximum number of results per page = 50. "This is an Asynchronous API. If you get a 202 code in response, please try again after 5 minutes." `Args:` airmeet_id: string The id of the Airmeet. `Returns:` Parsons.Table List of attendees for this Airmeet event """ return self._get_all_pages(url=f"airmeet/{airmeet_id}/attendees", page_size=50)
[docs] def fetch_session_attendance(self, session_id) -> Table: """ Get all attendees for a specific Airmeet session, handling pagination based on the returned pageCount. Results are available only for sessions with a status of `FINISHED`. Maximum number of results per page = 50. "This is an Asynchronous API. If you get a 202 code in response, please try again after 5 minutes." `Args:` session_id: string The id of the session. `Returns:` Parsons.Table List of attendees for this session """ return self._get_all_pages(url=f"session/{session_id}/attendees", page_size=50)
[docs] def fetch_airmeet_booths(self, airmeet_id) -> Table: """ Get the list of booths for a specific Airmeet by ID. `CAUTION: This method is untested. Booths are available only in certain Airmeet plans.` `Args:` airmeet_id: string The id of the Airmeet. `Returns:` Parsons.Table List of booths for this Airmeet """ response = self.client.get_request(url=f"airmeet/{airmeet_id}/booths") return Table(response["booths"] or [])
[docs] def fetch_booth_attendance(self, airmeet_id, booth_id) -> Table: """ Get all attendees for a specific Airmeet booth, handling pagination based on the returned pageCount. Results are available only for events with a status of `FINISHED`. Maximum number of results per page = 50. "This is an Asynchronous API. If you get a 202 code in response, please try again after 5 minutes." `CAUTION: This method is untested. Booths are available only in certain Airmeet plans.` `Args:` airmeet_id: string The id of the Airmeet. booth_id: string The id of the booth. `Returns:` Parsons.Table List of attendees for this booth """ return self._get_all_pages( url=f"airmeet/{airmeet_id}/booth/{booth_id}/booth-attendance", page_size=50 )
[docs] def fetch_poll_responses(self, airmeet_id) -> Table: """ Get a list of the poll responses in an Airmeet, handling pagination based on the returned pageCount. Maximum number of results per page = 50. `Args:` airmeet_id: string The id of the Airmeet. `Returns:` Parsons.Table List of users. For each user, the value for the "polls" key is a list of poll questions and answers for that user. """ return self._get_all_pages(url=f"airmeet/{airmeet_id}/polls", page_size=50)
[docs] def fetch_questions_asked(self, airmeet_id) -> Table: """ Get a list of the questions asked in an Airmeet. `Args:` airmeet_id: string The id of the Airmeet. `Returns:` Parsons.Table List of users. For each user, the value for the "questions" key is a list of that user's questions. """ response = self.client.get_request(url=f"airmeet/{airmeet_id}/questions") return Table(response["data"])
[docs] def fetch_event_tracks(self, airmeet_id) -> Table: """ Get a list of the tracks in a specific Airmeet by ID. `CAUTION: This method is untested. Event tracks are available only in certain Airmeet plans.` `Args:` airmeet_id: string The id of the Airmeet. `Returns:` Parsons.Table List of event tracks """ response = self.client.get_request(url=f"airmeet/{airmeet_id}/tracks") return Table(response["tracks"])
[docs] def fetch_registration_utms(self, airmeet_id) -> Table: """ Get all the UTM parameters captured during registration, handling pagination based on the returned pageCount. Maximum number of results per page = ?? (documentation doesn't say, but assume 50 like the other asynchronous APIs). "This is an Asynchronous API. If you get a 202 code in response, please try again after 5 minutes." `Args:` airmeet_id: string The id of the Airmeet. `Returns:` Parsons.Table List of UTM parameters captured during registration """ return self._get_all_pages(url=f"airmeet/{airmeet_id}/utms", page_size=50)
[docs] def download_session_recordings(self, airmeet_id, session_id=None) -> Table: """ Get a list of recordings for a specific Airmeet (and optionally a specific session in that Airmeet). The data for each recording includes a download link which is valid for 6 hours. The API returns "recordingsCount" and "totalCount", which implies that the results could be paged like in fetch_airmeet_participants(). The API docs don't specify if that's the case, but this method will need to be updated if it is. `Args:` airmeet_id: string The id of the Airmeet. session_id: string (optional) If provided, limits results to only the recording of the specified session. `Returns:` Parsons.Table List of session recordings """ kwargs = {} if session_id: kwargs["sessionIds"] = session_id response = self.client.get_request(url=f"airmeet/{airmeet_id}/session-recordings", **kwargs) return Table(response["recordings"])
[docs] def fetch_event_replay_attendance(self, airmeet_id, session_id=None) -> Table: """ Get all replay attendees for a specific Airmeet (and optionally a specific session in that Airmeet), handling pagination based on the returned pageCount. Results are available only for events with a status of `FINISHED`. Maximum number of results per page = 50. "This is an Asynchronous API. If you get a 202 code in response, please try again after 5 minutes." `Args:` airmeet_id: string The id of the Airmeet. session_id: string (optional) If provided, limits results to only attendees of the specified session. `Returns:` Parsons.Table List of event replay attendees """ attendees = self._get_all_pages( url=f"airmeet/{airmeet_id}/event-replay-attendees", page_size=50 ) if session_id is not None: attendees = attendees.select_rows("{session_id} == '" + session_id + "'") return attendees