wzlight
is an asynchronous Python wrapper for the Call of Duty API
that focuses on Warzone endpoints.
- Asynchronous with help of HTTPX, the HTTP client library
- Light : only centered around the few GET methods to collect Warzone stats
- Handle SSO auth. the (now) only available way to connect to the API
# with pip
pip install wzlight
# with Poetry
poetry add wzlight
import os
import asyncio
from pprint import pprint
import httpx
from dotenv import load_dotenv
from wzlight import Api
async def main():
load_dotenv()
sso = os.environ["SSO"]
username = "amadevs#1689"
platform = "battle"
# Initialize Api with a SSO token
# SSO token can be found in browser while login-in to callofduty.com
api = Api(sso)
# Wrapper internal methods are using the httpx async HTTP library
# Initially httpx.AsyncClient) was built in its own cls, but letting users define it at app-level --as a context manager,
# ensures more robustness and flexibility (custom timeout, limits...) and prevent async loop event errors.
# Thus, your main() should follow the preferred approach while working with an async http client:
async with httpx.AsyncClient() as httpxClient:
profile = await api.GetProfile(httpxClient, platform, username)
pprint(profile, depth=2)
recent_matches = await api.GetRecentMatches(httpxClient, platform, username)
recent_matches_short = [match for match in recent_matches[:2]]
pprint(recent_matches_short, depth=3)
matchStats = await api.GetMatch(
httpxClient, platform, matchId=11378702801403672847
)
pprint(matchStats, depth=1)
if __name__ == "__main__":
asyncio.run(main())
import os
import asyncio
from pprint import pprint
import httpx
from dotenv import load_dotenv
from typing import AsyncContextManager
from wzlight import Api
async def main():
load_dotenv()
sso = os.environ["SSO"]
username = "amadevs#1689"
platform = "battle"
# Initialize Api with a SSO token
api = Api(sso)
# We're running concurrently on match details endpoint, given a list of matchIds
# COD API is unofficial and undocumented, rate limits / restrictions are not known
# So you should probably --at least, handle some concurrency limits e.g for fetching a batch of matchs
# See. httpx.Limits, async.Semaphore or other librairies such as aiometer
matchIds = [
5850171651963062771,
6910618934945378397,
16975576559940046894,
639235311963231866,
11887968911271282782,
7897970481732864368,
]
# enclose api.getMatch into safe_GetMatch that accept a Semaphore threshold arg.
limit = asyncio.Semaphore(2)
async def safe_GetMatch(httpxClient, platform, matchId, sema: AsyncContextManager):
# No more than two concurrent tasks. If reaches two, wait 1 sec
async with sema:
r = await api.GetMatch(httpxClient, platform, matchId)
if sema.locked():
print("Concurrency limit reached, waiting ...")
await asyncio.sleep(1)
return r
# Do not forget to open Client as context manager :
async with httpx.AsyncClient() as httpxClient:
sema = asyncio.Semaphore(2)
tasks = []
for matchId in matchIds:
tasks.append(safe_GetMatch(httpxClient, platform, matchId, sema))
results = await asyncio.gather(*tasks, return_exceptions=True)
return resultss
print(len(results))
if __name__ == "__main__":
asyncio.run(main())
Inspiration (heavily) came from :
Also check those links if your need documentation on how the API works
- EthanC/CallofDuty.py : the most complete but now slightly deprecated (mainly the Auth.), async COD client (lot of exploited endpoints and methods + more than WZ)
- Lierrmm/Node-CallOfDuty : very clean async. wrapper written in NodeJS. Also check their Discord to get a grip on API subtleties and unofficial changes (privacy changes, rate limits etc)
- valtov/WarzoneStats : very clean synch. Python wrapper by the creator of wzstats.gg