Created
June 26, 2022 01:55
-
-
Save barrownicholas/610314cbf5d96d5aab191507f58930e9 to your computer and use it in GitHub Desktop.
Connect to the WaveApps API and run GraphQL queries.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from __future__ import annotations | |
# install requirements: pip3 install gql[all] | |
from gql import gql, Client | |
from gql.transport.aiohttp import AIOHTTPTransport | |
ENDPOINT = "https://gql.waveapps.com/graphql/public" # this is a static endpoint and should not be changed | |
class WaveApps: | |
def __init__(self, API_TOKEN) -> None: | |
# Select your transport with a defined url endpoint | |
self.transport = AIOHTTPTransport(url=ENDPOINT, headers={'Authorization': 'Bearer ' + API_TOKEN}) | |
# Create a GraphQL client using the defined transport | |
self.client = Client(transport=self.transport, fetch_schema_from_transport=True) | |
# Create the 'query' object for the WaveApps.query.{{ENDPOINT}} syntax | |
self.query = self.Query(self) | |
class Query: | |
def __init__(self, wave_connection: WaveApps) -> None: | |
self.wave = wave_connection | |
def custom(self, query_str: str, **kwargs) -> dict: | |
"""Run a custom query based on a query string; pass names query parameters in the query_str as kwargs. | |
:param query_str: a WaveApps-compatible query-string, such as: 'query { ... }' | |
:type query_str: str | |
:return: the result of the query | |
:rtype: dict | |
""" | |
query = gql(query_str) | |
result: dict = self.wave.client.execute(query, variable_values=kwargs) | |
return result | |
def user(self) -> dict: | |
# Provide a GraphQL query | |
query = gql( | |
""" | |
query { | |
user { | |
id | |
firstName | |
lastName | |
defaultEmail | |
createdAt | |
modifiedAt | |
} | |
} | |
""" | |
) | |
return self.wave.client.execute(query) | |
def businesses(self, page: int = 1, page_size: int = 10, list_only = True) -> list | dict: | |
"""Query a list of businesses. If you are unsure if you have retrieved all items, set list_only=False for additional output. | |
:param page: the page to return, defaults to 1 | |
:type page: int, optional | |
:param page_size: the size of each page, defaults to 10 | |
:type page_size: int, optional | |
:param list_only: return only the list of businesses from the entire output, defaults to True | |
:type list_only: bool, optional | |
:return: either the list of businesses or the entire dict result, depending on 'list_only' | |
:rtype: list | dict | |
""" | |
query = gql(f""" | |
query {{ | |
businesses(page: {page}, pageSize: {page_size}) {{ | |
pageInfo {{ | |
currentPage | |
totalPages | |
totalCount | |
}} | |
edges {{ | |
node {{ | |
id | |
name | |
isClassicAccounting | |
isClassicInvoicing | |
isPersonal | |
}} | |
}} | |
}} | |
}} | |
""" | |
) | |
result: dict = self.wave.client.execute(query) | |
if list_only: | |
filtered = result['businesses']['edges'] | |
return filtered | |
else: | |
return result | |
def invoices(self, business_id: str, page: int = 1, page_size: int = 40, list_only = True) -> list | dict: | |
"""Query a list of invoices from a business. If you are unsure if you have retrieved all items, set list_only=False for additional output. | |
:param business_id: the ID of the business to retrieve invoices for (use `WaveApps.query.businesses` to obtain it) | |
:type business_id: str | |
:param page: the page to return, defaults to 1 | |
:type page: int, optional | |
:param page_size: the size of each page, defaults to 20 | |
:type page_size: int, optional | |
:param list_only: return only the list of businesses from the entire output, defaults to True | |
:type list_only: bool, optional | |
:return: either the list of businesses or the entire dict result, depending on 'list_only' | |
:rtype: list | dict | |
""" | |
# TODO: fix the 'return_all' component | |
params = { | |
"businessId": f"{business_id}", | |
"page": page, | |
"pageSize": page_size | |
} | |
query = gql(f""" | |
query($businessId: ID!, $page: Int!, $pageSize: Int!) {{ | |
business(id: $businessId) {{ | |
id | |
isClassicInvoicing | |
invoices(page: $page, pageSize: $pageSize) {{ | |
pageInfo {{ | |
currentPage | |
totalPages | |
totalCount | |
}} | |
edges {{ | |
node {{ | |
id | |
createdAt | |
modifiedAt | |
pdfUrl | |
viewUrl | |
status | |
title | |
subhead | |
invoiceNumber | |
invoiceDate | |
poNumber | |
customer {{ | |
id | |
name | |
# Can add additional customer fields here | |
}} | |
currency {{ | |
code | |
}} | |
dueDate | |
amountDue {{ | |
value | |
currency {{ | |
symbol | |
}} | |
}} | |
amountPaid {{ | |
value | |
currency {{ | |
symbol | |
}} | |
}} | |
taxTotal {{ | |
value | |
currency {{ | |
symbol | |
}} | |
}} | |
total {{ | |
value | |
currency {{ | |
symbol | |
}} | |
}} | |
exchangeRate | |
footer | |
memo | |
disableCreditCardPayments | |
disableBankPayments | |
itemTitle | |
unitTitle | |
priceTitle | |
amountTitle | |
hideName | |
hideDescription | |
hideUnit | |
hidePrice | |
hideAmount | |
items {{ | |
product {{ | |
id | |
name | |
# Can add additional product fields here | |
}} | |
description | |
quantity | |
price | |
subtotal {{ | |
value | |
currency {{ | |
symbol | |
}} | |
}} | |
total {{ | |
value | |
currency {{ | |
symbol | |
}} | |
}} | |
account {{ | |
id | |
name | |
subtype {{ | |
name | |
value | |
}} | |
# Can add additional account fields here | |
}} | |
taxes {{ | |
amount {{ | |
value | |
}} | |
salesTax {{ | |
id | |
name | |
# Can add additional sales tax fields here | |
}} | |
}} | |
}} | |
lastSentAt | |
lastSentVia | |
lastViewedAt | |
}} | |
}} | |
}} | |
}} | |
}} | |
""") | |
result: dict = self.wave.client.execute(query, variable_values=params) | |
if list_only: | |
filtered = result['business']['invoices']['edges'] | |
return filtered | |
else: | |
return result |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment