Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
logger.info(
f"Requesting run of {query} at {connection.url}/api/{connection.api_version}"
)
r = connection.post_json(route="run", data=query)
if r.status_code == 202:
query_id = r.headers["Location"].split("/").pop()
logger.info(
f"Accepted {query} at {connection.url}/api/{connection.api_version} with id {query_id}"
)
return query_id
else:
try:
error = r.json()["msg"]
except (ValueError, KeyError):
error = "Unknown error."
raise FlowclientConnectionError(
f"Error running the query: {error}. Status code: {r.status_code}."
)
elif response.status_code in {401, 403}:
try:
error = response.json()["msg"]
except (ValueError, KeyError):
error = "Unknown access denied error"
raise FlowclientConnectionError(error)
else:
try:
error = response.json()["msg"]
except (ValueError, KeyError):
error = "Unknown error"
try:
status = response.json()["status"]
except (ValueError, KeyError):
status = "Unknown status"
raise FlowclientConnectionError(
f"Something went wrong: {error}. API returned with status code: {response.status_code} and status '{status}'"
)
else:
try:
error_msg = response.json()["msg"]
try:
returned_payload = response.json()["payload"]
payload_info = (
"" if not returned_payload else f" Payload: {returned_payload}"
)
except KeyError:
payload_info = ""
except ValueError:
# Happens if the response body does not contain valid JSON
# (see http://docs.python-requests.org/en/master/api/#requests.Response.json)
error_msg = f"the response did not contain valid JSON"
payload_info = ""
raise FlowclientConnectionError(
f"Something went wrong. API returned with status code {response.status_code}. Error message: '{error_msg}'.{payload_info}"
)
Returns
-------
pandas.DataFrame
Dataframe containing the result
"""
response = connection.get_url(route=location)
if response.status_code != 200:
try:
msg = response.json()["msg"]
more_info = f" Reason: {msg}"
except KeyError:
more_info = ""
raise FlowclientConnectionError(
f"Could not get result. API returned with status code: {response.status_code}.{more_info}"
)
result = response.json()
logger.info(f"Got {connection.url}/api/{connection.api_version}/{location}")
return pd.DataFrame.from_records(result["query_result"])
except ConnectionError as e:
error_msg = f"Unable to connect to FlowKit API at {self.url}: {e}"
logger.info(error_msg)
raise FlowclientConnectionError(error_msg)
if response.status_code in {202, 200, 303}:
return response
elif response.status_code == 404:
raise FileNotFoundError(
f"{self.url}/api/{self.api_version}/{route} not found."
)
elif response.status_code in {401, 403}:
try:
error = response.json()["msg"]
except (ValueError, KeyError):
error = "Unknown access denied error"
raise FlowclientConnectionError(error)
else:
try:
error = response.json()["msg"]
except (ValueError, KeyError):
error = "Unknown error"
try:
status = response.json()["status"]
except (ValueError, KeyError):
status = "Unknown status"
raise FlowclientConnectionError(
f"Something went wrong: {error}. API returned with status code: {response.status_code} and status '{status}'"
)
-------
dict
Available dates in the format {event_type: [list of dates]}
"""
logger.info(
f"Getting {connection.url}/api/{connection.api_version}/available_dates"
)
response = connection.get_url(route=f"available_dates")
if response.status_code != 200:
try:
msg = response.json()["msg"]
more_info = f" Reason: {msg}"
except KeyError:
more_info = ""
raise FlowclientConnectionError(
f"Could not get available dates. API returned with status code: {response.status_code}.{more_info}"
)
result = response.json()["available_dates"]
logger.info(f"Got {connection.url}/api/{connection.api_version}/available_dates")
if event_types is None:
return result
else:
return {k: v for k, v in result.items() if k in event_types}
Identifier of the query to retrieve
Returns
-------
str
"Finished" or "Running"
"""
ready, reply = query_is_ready(connection=connection, query_id=query_id)
if ready:
return "Finished"
else:
try:
return reply.json()["status"]
except (KeyError, TypeError):
raise FlowclientConnectionError(f"No status reported.")
"""
logger.info(
f"Polling server on {connection.url}/api/{connection.api_version}/poll/{query_id}"
)
reply = connection.get_url(route=f"poll/{query_id}")
if reply.status_code == 303:
logger.info(
f"{connection.url}/api/{connection.api_version}/poll/{query_id} ready."
)
return True, reply # Query is ready, so exit the loop
elif reply.status_code == 202:
return False, reply
else:
raise FlowclientConnectionError(
f"Something went wrong: {reply}. API returned with status code: {reply.status_code}"
)
Returns
-------
requests.Response
"""
logger.debug(f"Getting {self.url}/api/{self.api_version}/{route}")
try:
response = self.session.get(
f"{self.url}/api/{self.api_version}/{route}",
allow_redirects=False,
json=data,
)
except ConnectionError as e:
error_msg = f"Unable to connect to FlowKit API at {self.url}: {e}"
logger.info(error_msg)
raise FlowclientConnectionError(error_msg)
if response.status_code in {202, 200, 303}:
return response
elif response.status_code == 404:
raise FileNotFoundError(
f"{self.url}/api/{self.api_version}/{route} not found."
)
elif response.status_code in {401, 403}:
try:
error = response.json()["msg"]
except (ValueError, KeyError):
error = "Unknown access denied error"
raise FlowclientConnectionError(error)
else:
try:
error = response.json()["msg"]
except (ValueError, KeyError):