aiohttp powered apollo like graphql client
Project description
rath
DEVELOPMENT
Inspiration
Rath is like Apollo, but for python. It adheres to the design principle of Links and enables complex GraphQL setups, like seperation of query and subscription endpoints, dynamic token loading, etc..
Installation
pip install rath
Usage Example
from rath.links.auth import AuthTokenLink
from rath.links.aiohttp import AioHttpLink
from rath.links import compose, split
from rath.gql import gql
async def aload_token():
return "SERVER_TOKEN"
auth = AuthTokenLink(token_loader=aload_token)
link = AioHttpLink(url="https://api.spacex.land/graphql/")
rath = Rath(links=compose(auth,link))
rath.connect()
query = """query TestQuery {
capsules {
id
missions {
flight
}
}
}
"""
result = rath.execute(query)
This example composes both the AuthToken and AioHttp link: During each query the Bearer headers are set to the retrieved token, on authentication fail (for example if Token Expired) the AuthToken automatically refetches the token and retries the query.
Async Usage
Rath is build with koil, for async/sync compatibility but also exposed a complete asynhronous api
from rath.links.auth import AuthTokenLink
from rath.links.aiohttp import AioHttpLink
from rath.links import compose, split
from rath.gql import gql
async def aload_token():
return "SERVER_TOKEN"
auth = AuthTokenLink(token_loader=aload_token)
link = AioHttpLink(url="https://api.spacex.land/graphql/")
async def main():
rath = Rath(links=compose(auth,link))
await rath.aconnect()
query = """query TestQuery {
capsules {
id
missions {
flight
}
}
}
"""
result = await rath.aexecute(query)
asyncio.run(main())
Included Links
- Reconnecting WebsocketLink (untested)
- AioHttpLink (supports multipart uploads)
- SplitLink (allows to split the terminating link - Subscription into WebsocketLink, Query, Mutation into Aiohttp)
- AuthTokenLink (Token insertion with automatic refresh)
Why Rath
Well "apollo" is already taken as a name, and rath (according to wikipedia) is an etruscan deity identified with Apollo.
Rath + Turms
Rath works especially well with turms generated typed operations:
import asyncio
from examples.api.schema import aget_capsules
from rath.rath import Rath
from rath.links.aiohttp import AIOHttpLink
from rath.links.auth import AuthTokenLink
from rath.links.compose import compose
async def token_loader():
return ""
link = compose(
AuthTokenLink(token_loader), AIOHttpLink("https://api.spacex.land/graphql/")
)
RATH = Rath(
link=link,
register=True, # allows global access (singleton-antipattern, but rath has no state)
)
async def main():
capsules = await aget_capsules() # fully typed pydantic powered dataclasses generated through turms
print(capsules)
asyncio.run(main())
Examples
This github repository also contains an example client with a turms generated query with the public SpaceX api, as well as a sample of the generated api.
Parsers
Besides links, there is also support for sequentially parsing of the Operation before it enters the asynchronous links (example in thread upload to s3fs), this is aided through parsers that provide both async and sync interfaces.
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
