22 KiB
22 KiB
Python coding
We used openapi-python-client to generate client libraries for the beatleader.xyz api. It's in clients/beatleader in our python project:
.
├── docs/
│ ├── prompts/
│ └── *.md
├── src/
│ ├── clients/
│ │ ├── beatleader/
│ │ │ ├── api/ (various API endpoints)
│ │ │ ├── models/ (data models)
│ │ │ └── client.py, errors.py, __init__.py, types.py
│ │ ├── beatsaver/ (similar structure to beatleader)
│ │ └── scoresaber/ (similar structure to beatleader)
│ ├── helpers/
│ │ └── *.py
│ └── saberlist/
│ └── *.py
├── tests/
│ ├── assets/
│ └── playlist_builder.py
├── pyproject.toml
└── README.md
Here's the clients/beatleader dir:
treegit src/clients/beatleader/
src/clients/beatleader/
├── api
│ ├── beast_saber
│ │ ├── beast_saber_get_all.py
│ │ ├── beast_saber_nominate.py
│ │ └── __init__.py
│ ├── clan
│ │ ├── clan_get_all.py
│ │ ├── clan_get_clan_by_id.py
│ │ ├── clan_get_clan.py
│ │ ├── clan_get_clan_with_maps_by_id.py
│ │ ├── clan_get_clan_with_maps.py
│ │ ├── clan_get_history.py
│ │ ├── clan_global_map.py
│ │ └── __init__.py
│ ├── leaderboard
│ │ ├── __init__.py
│ │ ├── leaderboard_get_all.py
│ │ ├── leaderboard_get_clan_rankings.py
│ │ ├── leaderboard_get.py
│ │ └── leaderboard_get_scoregraph.py
│ ├── modifiers
│ │ ├── __init__.py
│ │ └── modifiers_get_modifiers.py
│ ├── patreon
│ │ ├── __init__.py
│ │ └── patreon_refresh_my_patreon.py
│ ├── player
│ │ ├── __init__.py
│ │ ├── player_get_beat_saver.py
│ │ ├── player_get_discord.py
│ │ ├── player_get_followers_info.py
│ │ ├── player_get_followers.py
│ │ ├── player_get_founded_clan.py
│ │ ├── player_get_participating_events.py
│ │ ├── player_get_patreon.py
│ │ ├── player_get_players.py
│ │ ├── player_get.py
│ │ └── player_get_ranked_maps.py
│ ├── player_scores
│ │ ├── __init__.py
│ │ ├── player_scores_acc_graph.py
│ │ ├── player_scores_get_compact_history.py
│ │ ├── player_scores_get_compact_scores.py
│ │ ├── player_scores_get_history.py
│ │ ├── player_scores_get_pinned_scores.py
│ │ ├── player_scores_get_scores.py
│ │ └── player_scores_get_score_value.py
│ ├── song
│ │ ├── __init__.py
│ │ └── song_get_all.py
│ └── __init__.py
├── models
│ ├── __pycache__
│ ├── achievement_description.py
│ ├── achievement_level.py
│ ├── achievement.py
│ ├── badge.py
│ ├── ban.py
│ ├── beasties_nomination.py
│ ├── besties_nomination_response.py
│ ├── clan_bigger_response.py
│ ├── clan_global_map_point.py
│ ├── clan_global_map.py
│ ├── clan_map_connection.py
│ ├── clan_maps_sort_by.py
│ ├── clan_point.py
│ ├── clan.py
│ ├── clan_ranking_response_clan_response_full_response_with_metadata_and_container.py
│ ├── clan_ranking_response.py
│ ├── clan_response_full.py
│ ├── clan_response_full_response_with_metadata.py
│ ├── clan_response.py
│ ├── clan_sort_by.py
│ ├── compact_leaderboard.py
│ ├── compact_leaderboard_response.py
│ ├── compact_score.py
│ ├── compact_score_response.py
│ ├── compact_score_response_response_with_metadata.py
│ ├── compact_song_response.py
│ ├── controller_enum.py
│ ├── criteria_commentary.py
│ ├── difficulty_description.py
│ ├── difficulty_response.py
│ ├── difficulty_status.py
│ ├── event_player.py
│ ├── event_ranking.py
│ ├── external_status.py
│ ├── featured_playlist.py
│ ├── featured_playlist_response.py
│ ├── follower_type.py
│ ├── global_map_history.py
│ ├── history_compact_response.py
│ ├── hmd.py
│ ├── info_to_highlight.py
│ ├── __init__.py
│ ├── leaderboard_change.py
│ ├── leaderboard_clan_ranking_response.py
│ ├── leaderboard_contexts.py
│ ├── leaderboard_group_entry.py
│ ├── leaderboard_info_response.py
│ ├── leaderboard_info_response_response_with_metadata.py
│ ├── leaderboard.py
│ ├── leaderboard_response.py
│ ├── leaderboard_sort_by.py
│ ├── legacy_modifiers.py
│ ├── link_response.py
│ ├── map_diff_response.py
│ ├── map_info_response.py
│ ├── map_info_response_response_with_metadata.py
│ ├── mapper.py
│ ├── mapper_response.py
│ ├── map_quality.py
│ ├── map_sort_by.py
│ ├── maps_type.py
│ ├── metadata.py
│ ├── modifiers_map.py
│ ├── modifiers_rating.py
│ ├── my_type.py
│ ├── operation.py
│ ├── order.py
│ ├── participating_event_response.py
│ ├── patreon_features.py
│ ├── player_change.py
│ ├── player_context_extension.py
│ ├── player_follower.py
│ ├── player_followers_info_response.py
│ ├── player.py
│ ├── player_response_clan_response_full_response_with_metadata_and_container.py
│ ├── player_response_full.py
│ ├── player_response.py
│ ├── player_response_with_stats.py
│ ├── player_response_with_stats_response_with_metadata.py
│ ├── player_score_stats_history.py
│ ├── player_score_stats.py
│ ├── player_search.py
│ ├── player_social.py
│ ├── player_sort_by.py
│ ├── pp_type.py
│ ├── profile_settings.py
│ ├── qualification_change.py
│ ├── qualification_commentary.py
│ ├── qualification_vote.py
│ ├── ranked_mapper_response.py
│ ├── ranked_map.py
│ ├── rank_qualification.py
│ ├── rank_update_change.py
│ ├── rank_update.py
│ ├── rank_voting.py
│ ├── replay_offsets.py
│ ├── requirements.py
│ ├── score_filter_status.py
│ ├── score_graph_entry.py
│ ├── score_improvement.py
│ ├── score_metadata.py
│ ├── score_response.py
│ ├── score_response_with_acc.py
│ ├── score_response_with_my_score.py
│ ├── score_response_with_my_score_response_with_metadata.py
│ ├── scores_sort_by.py
│ ├── song.py
│ ├── song_response.py
│ ├── song_status.py
│ ├── type.py
│ └── voter_feedback.py
├── __pycache__
├── client.py
├── errors.py
├── __init__.py
├── py.typed
└── types.py
13 directories, 158 files
Here's the contents of src/clients/beatleader/client.py:
import ssl
from typing import Any, Dict, Optional, Union
import httpx
from attrs import define, evolve, field
@define
class Client:
"""A class for keeping track of data related to the API
The following are accepted as keyword arguments and will be used to construct httpx Clients internally:
``base_url``: The base URL for the API, all requests are made to a relative path to this URL
``cookies``: A dictionary of cookies to be sent with every request
``headers``: A dictionary of headers to be sent with every request
``timeout``: The maximum amount of a time a request can take. API functions will raise
httpx.TimeoutException if this is exceeded.
``verify_ssl``: Whether or not to verify the SSL certificate of the API server. This should be True in production,
but can be set to False for testing purposes.
``follow_redirects``: Whether or not to follow redirects. Default value is False.
``httpx_args``: A dictionary of additional arguments to be passed to the ``httpx.Client`` and ``httpx.AsyncClient`` constructor.
Attributes:
raise_on_unexpected_status: Whether or not to raise an errors.UnexpectedStatus if the API returns a
status code that was not documented in the source OpenAPI document. Can also be provided as a keyword
argument to the constructor.
"""
raise_on_unexpected_status: bool = field(default=False, kw_only=True)
_base_url: str = field(alias="base_url")
_cookies: Dict[str, str] = field(factory=dict, kw_only=True, alias="cookies")
_headers: Dict[str, str] = field(factory=dict, kw_only=True, alias="headers")
_timeout: Optional[httpx.Timeout] = field(default=None, kw_only=True, alias="timeout")
_verify_ssl: Union[str, bool, ssl.SSLContext] = field(default=True, kw_only=True, alias="verify_ssl")
_follow_redirects: bool = field(default=False, kw_only=True, alias="follow_redirects")
_httpx_args: Dict[str, Any] = field(factory=dict, kw_only=True, alias="httpx_args")
_client: Optional[httpx.Client] = field(default=None, init=False)
_async_client: Optional[httpx.AsyncClient] = field(default=None, init=False)
def with_headers(self, headers: Dict[str, str]) -> "Client":
"""Get a new client matching this one with additional headers"""
if self._client is not None:
self._client.headers.update(headers)
if self._async_client is not None:
self._async_client.headers.update(headers)
return evolve(self, headers={**self._headers, **headers})
def with_cookies(self, cookies: Dict[str, str]) -> "Client":
"""Get a new client matching this one with additional cookies"""
if self._client is not None:
self._client.cookies.update(cookies)
if self._async_client is not None:
self._async_client.cookies.update(cookies)
return evolve(self, cookies={**self._cookies, **cookies})
def with_timeout(self, timeout: httpx.Timeout) -> "Client":
"""Get a new client matching this one with a new timeout (in seconds)"""
if self._client is not None:
self._client.timeout = timeout
if self._async_client is not None:
self._async_client.timeout = timeout
return evolve(self, timeout=timeout)
def set_httpx_client(self, client: httpx.Client) -> "Client":
"""Manually the underlying httpx.Client
**NOTE**: This will override any other settings on the client, including cookies, headers, and timeout.
"""
self._client = client
return self
def get_httpx_client(self) -> httpx.Client:
"""Get the underlying httpx.Client, constructing a new one if not previously set"""
if self._client is None:
self._client = httpx.Client(
base_url=self._base_url,
cookies=self._cookies,
headers=self._headers,
timeout=self._timeout,
verify=self._verify_ssl,
follow_redirects=self._follow_redirects,
**self._httpx_args,
)
return self._client
def __enter__(self) -> "Client":
"""Enter a context manager for self.client—you cannot enter twice (see httpx docs)"""
self.get_httpx_client().__enter__()
return self
def __exit__(self, *args: Any, **kwargs: Any) -> None:
"""Exit a context manager for internal httpx.Client (see httpx docs)"""
self.get_httpx_client().__exit__(*args, **kwargs)
def set_async_httpx_client(self, async_client: httpx.AsyncClient) -> "Client":
"""Manually the underlying httpx.AsyncClient
**NOTE**: This will override any other settings on the client, including cookies, headers, and timeout.
"""
self._async_client = async_client
return self
def get_async_httpx_client(self) -> httpx.AsyncClient:
"""Get the underlying httpx.AsyncClient, constructing a new one if not previously set"""
if self._async_client is None:
self._async_client = httpx.AsyncClient(
base_url=self._base_url,
cookies=self._cookies,
headers=self._headers,
timeout=self._timeout,
verify=self._verify_ssl,
follow_redirects=self._follow_redirects,
**self._httpx_args,
)
return self._async_client
async def __aenter__(self) -> "Client":
"""Enter a context manager for underlying httpx.AsyncClient—you cannot enter twice (see httpx docs)"""
await self.get_async_httpx_client().__aenter__()
return self
async def __aexit__(self, *args: Any, **kwargs: Any) -> None:
"""Exit a context manager for underlying httpx.AsyncClient (see httpx docs)"""
await self.get_async_httpx_client().__aexit__(*args, **kwargs)
@define
class AuthenticatedClient:
"""A Client which has been authenticated for use on secured endpoints
The following are accepted as keyword arguments and will be used to construct httpx Clients internally:
``base_url``: The base URL for the API, all requests are made to a relative path to this URL
``cookies``: A dictionary of cookies to be sent with every request
``headers``: A dictionary of headers to be sent with every request
``timeout``: The maximum amount of a time a request can take. API functions will raise
httpx.TimeoutException if this is exceeded.
``verify_ssl``: Whether or not to verify the SSL certificate of the API server. This should be True in production,
but can be set to False for testing purposes.
``follow_redirects``: Whether or not to follow redirects. Default value is False.
``httpx_args``: A dictionary of additional arguments to be passed to the ``httpx.Client`` and ``httpx.AsyncClient`` constructor.
Attributes:
raise_on_unexpected_status: Whether or not to raise an errors.UnexpectedStatus if the API returns a
status code that was not documented in the source OpenAPI document. Can also be provided as a keyword
argument to the constructor.
token: The token to use for authentication
prefix: The prefix to use for the Authorization header
auth_header_name: The name of the Authorization header
"""
raise_on_unexpected_status: bool = field(default=False, kw_only=True)
_base_url: str = field(alias="base_url")
_cookies: Dict[str, str] = field(factory=dict, kw_only=True, alias="cookies")
_headers: Dict[str, str] = field(factory=dict, kw_only=True, alias="headers")
_timeout: Optional[httpx.Timeout] = field(default=None, kw_only=True, alias="timeout")
_verify_ssl: Union[str, bool, ssl.SSLContext] = field(default=True, kw_only=True, alias="verify_ssl")
_follow_redirects: bool = field(default=False, kw_only=True, alias="follow_redirects")
_httpx_args: Dict[str, Any] = field(factory=dict, kw_only=True, alias="httpx_args")
_client: Optional[httpx.Client] = field(default=None, init=False)
_async_client: Optional[httpx.AsyncClient] = field(default=None, init=False)
token: str
prefix: str = "Bearer"
auth_header_name: str = "Authorization"
def with_headers(self, headers: Dict[str, str]) -> "AuthenticatedClient":
"""Get a new client matching this one with additional headers"""
if self._client is not None:
self._client.headers.update(headers)
if self._async_client is not None:
self._async_client.headers.update(headers)
return evolve(self, headers={**self._headers, **headers})
def with_cookies(self, cookies: Dict[str, str]) -> "AuthenticatedClient":
"""Get a new client matching this one with additional cookies"""
if self._client is not None:
self._client.cookies.update(cookies)
if self._async_client is not None:
self._async_client.cookies.update(cookies)
return evolve(self, cookies={**self._cookies, **cookies})
def with_timeout(self, timeout: httpx.Timeout) -> "AuthenticatedClient":
"""Get a new client matching this one with a new timeout (in seconds)"""
if self._client is not None:
self._client.timeout = timeout
if self._async_client is not None:
self._async_client.timeout = timeout
return evolve(self, timeout=timeout)
def set_httpx_client(self, client: httpx.Client) -> "AuthenticatedClient":
"""Manually the underlying httpx.Client
**NOTE**: This will override any other settings on the client, including cookies, headers, and timeout.
"""
self._client = client
return self
def get_httpx_client(self) -> httpx.Client:
"""Get the underlying httpx.Client, constructing a new one if not previously set"""
if self._client is None:
self._headers[self.auth_header_name] = f"{self.prefix} {self.token}" if self.prefix else self.token
self._client = httpx.Client(
base_url=self._base_url,
cookies=self._cookies,
headers=self._headers,
timeout=self._timeout,
verify=self._verify_ssl,
follow_redirects=self._follow_redirects,
**self._httpx_args,
)
return self._client
def __enter__(self) -> "AuthenticatedClient":
"""Enter a context manager for self.client—you cannot enter twice (see httpx docs)"""
self.get_httpx_client().__enter__()
return self
def __exit__(self, *args: Any, **kwargs: Any) -> None:
"""Exit a context manager for internal httpx.Client (see httpx docs)"""
self.get_httpx_client().__exit__(*args, **kwargs)
def set_async_httpx_client(self, async_client: httpx.AsyncClient) -> "AuthenticatedClient":
"""Manually the underlying httpx.AsyncClient
**NOTE**: This will override any other settings on the client, including cookies, headers, and timeout.
"""
self._async_client = async_client
return self
def get_async_httpx_client(self) -> httpx.AsyncClient:
"""Get the underlying httpx.AsyncClient, constructing a new one if not previously set"""
if self._async_client is None:
self._headers[self.auth_header_name] = f"{self.prefix} {self.token}" if self.prefix else self.token
self._async_client = httpx.AsyncClient(
base_url=self._base_url,
cookies=self._cookies,
headers=self._headers,
timeout=self._timeout,
verify=self._verify_ssl,
follow_redirects=self._follow_redirects,
**self._httpx_args,
)
return self._async_client
async def __aenter__(self) -> "AuthenticatedClient":
"""Enter a context manager for underlying httpx.AsyncClient—you cannot enter twice (see httpx docs)"""
await self.get_async_httpx_client().__aenter__()
return self
async def __aexit__(self, *args: Any, **kwargs: Any) -> None:
"""Exit a context manager for underlying httpx.AsyncClient (see httpx docs)"""
await self.get_async_httpx_client().__aexit__(*args, **kwargs)
Here is our attempt at using this client in ipython:
import json
import os
import logging
from datetime import datetime, timedelta
from typing import Optional, Dict, Any
from clients.beatleader import client as beatleader_client
from clients.beatleader.api.player_scores import player_scores_get_compact_scores
from clients.beatleader.models.score_response_with_my_score_response_with_metadata import ScoreResponseWithMyScoreResponseWithMetadata
logging.basicConfig(
format='%(asctime)s %(levelname)s: %(message)s',
datefmt='%Y-%m-%d %H:%M:%S',
level=logging.DEBUG
)
player_id = '76561199407393962'
BASE_URL = "https://api.beatleader.xyz"
client = beatleader_client.Client(base_url=BASE_URL)
response: ScoreResponseWithMyScoreResponseWithMetadata = player_scores_get_compact_scores.sync(
client=client,
id=player_id)
And the result:
---------------------------------------------------------------------------
AttributeError Traceback (most recent call last)
Cell In[1], line 21
17 player_id = '76561199407393962'
19 BASE_URL = "https://api.beatleader.xyz"
---> 21 response: ScoreResponseWithMyScoreResponseWithMetadata = player_scores_get_compact_scores.sync_detailed(
22 client=beatleader_client,
23 id=player_id)
File ~/ops/beatsaber/playlist-tool/src/clients/beatleader/api/player_scores/player_scores_get_compact_scores.py:216, in sync_detailed(id, client, sort_by, order, page, count, search, diff, mode, requirements, score_status, leaderboard_context, type, modifiers, stars_from, stars_to, time_from, time_to, event_id)
162 """Retrieve player's scores in a compact form
163
164 Fetches a paginated list of scores for a specified player ID. Returns less info to save bandwith or
(...)
192 Response[Union[Any, CompactScoreResponseResponseWithMetadata]]
193 """
195 kwargs = _get_kwargs(
196 id=id,
197 sort_by=sort_by,
(...)
213 event_id=event_id,
214 )
--> 216 response = client.get_httpx_client().request(
217 **kwargs,
218 )
220 return _build_response(client=client, response=response)
AttributeError: module 'clients.beatleader.client' has no attribute 'get_httpx_client'