API Reference¶
QueryKit¶
- class pnwkit.new.QueryKit(api_key: str, bot_key: Optional[str] = None, bot_key_api_key: Optional[str] = None, *, parse_int: Optional[Callable[[str], Any]] = None, parse_float: Optional[Callable[[str], Any]] = None, url: Optional[str] = None, socket_url: Optional[str] = None, subscription_url: Optional[str] = None, subscription_auth_url: Optional[str] = None, socket: Optional[Socket] = None, aiohttp_session: Optional[aiohttp.ClientSession] = None, requests_session: Optional[requests.Session] = None)¶
- mutation(field: MutationFieldLiteral, arguments: Dict[str, Union[Argument, Sequence[Argument]]], *fields: FieldValue, **variables: MutableMapping[str, Any]) → Mutation[Result]¶
Create a new mutation with this QueryKit.
- Parameters
field (RootFieldLiteral) – The subscription to query
arguments (Dict[
str
, Union[Argument, Sequence[Argument]]]) – The parameters to provide to the subscription to filter the eventsfields (Union[str,
Field
]) – The fields to fetch in the queryvariables (MutableMapping[
str
, Any]) – The values of any variables specified in the query
- Returns
A Mutation that can be subscribed too.
- Return type
Mutation[Any]
- query(field: RootFieldLiteral, arguments: Dict[str, Union[Argument, Sequence[Argument]]], *fields: FieldValue, **variables: MutableMapping[str, Any]) → Query[Result]¶
Create a new query with this QueryKit.
- Parameters
field (RootFieldLiteral) – The name of the root field to query
arguments (Dict[str, Union[Argument, Sequence[Argument]]]) – The parameters to provide to the field to filter results
fields (Union[str,
Field
]) – The fields to fetch in the queryvariables (MutableMapping[str, Any]) – The values of any variables specified in the query
- Returns
A Query that can be fetched or have additional queries called on it.
- Return type
- query_as(field: RootFieldLiteral, alias: str, arguments: Dict[str, Union[Argument, Sequence[Argument]]], *fields: FieldValue, **variables: MutableMapping[str, Any]) → Query[Result]¶
Create a new query with this QueryKit.
- Parameters
field (RootFieldLiteral) – The name of the root field to query
alias (
str
) – The name to have to results of this query returned by on the Resultarguments (Dict[
str
, Union[Argument, Sequence[Argument]]]) – The parameters to provide to the field to filter resultsfields (Union[str,
Field
]) – The fields to fetch in the queryvariables (MutableMapping[
str
, Any]) – The values of any variables specified in the query
- Returns
A Query that can be fetched or have additional queries called on it.
- Return type
- async subscribe(model: Literal[alliance_position], event: SubscriptionEventLiteral, filters: Optional[SubscriptionFilters] = None) → pnwkit.new.Subscription[pnwkit.data.AlliancePosition]¶
- async subscribe(model: Literal[bankrec], event: SubscriptionEventLiteral, filters: Optional[SubscriptionFilters] = None) → pnwkit.new.Subscription[pnwkit.data.Bankrec]
- async subscribe(model: Literal[bbgame], event: SubscriptionEventLiteral, filters: Optional[SubscriptionFilters] = None) → pnwkit.new.Subscription[pnwkit.data.BBGame]
- async subscribe(model: Literal[bbteam], event: SubscriptionEventLiteral, filters: Optional[SubscriptionFilters] = None) → pnwkit.new.Subscription[pnwkit.data.BBTeam]
- async subscribe(model: Literal[bounty], event: SubscriptionEventLiteral, filters: Optional[SubscriptionFilters] = None) → pnwkit.new.Subscription[pnwkit.data.Bounty]
- async subscribe(model: Literal[city], event: SubscriptionEventLiteral, filters: Optional[SubscriptionFilters] = None) → pnwkit.new.Subscription[pnwkit.data.City]
- async subscribe(model: Literal[nation], event: SubscriptionEventLiteral, filters: Optional[SubscriptionFilters] = None) → pnwkit.new.Subscription[pnwkit.data.Nation]
- async subscribe(model: Literal[tax_bracket], event: SubscriptionEventLiteral, filters: Optional[SubscriptionFilters] = None) → pnwkit.new.Subscription[pnwkit.data.TaxBracket]
- async subscribe(model: Literal[trade], event: SubscriptionEventLiteral, filters: Optional[SubscriptionFilters] = None) → pnwkit.new.Subscription[pnwkit.data.Trade]
- async subscribe(model: Literal[treaty], event: SubscriptionEventLiteral, filters: Optional[SubscriptionFilters] = None) → pnwkit.new.Subscription[pnwkit.data.Treaty]
- async subscribe(model: Literal[warattack], event: SubscriptionEventLiteral, filters: Optional[SubscriptionFilters] = None) → pnwkit.new.Subscription[pnwkit.data.WarAttack]
- async subscribe(model: Literal[war], event: SubscriptionEventLiteral, filters: Optional[SubscriptionFilters] = None) → pnwkit.new.Subscription[pnwkit.data.War]
- async subscribe(model: Literal[treasure_trade], event: SubscriptionEventLiteral, filters: Optional[SubscriptionFilters] = None) → pnwkit.new.Subscription[pnwkit.data.TreasureTrade]
- async subscribe(model: Literal[embargo], event: SubscriptionEventLiteral, filters: Optional[SubscriptionFilters] = None) → pnwkit.new.Subscription[pnwkit.data.Embargo]
- async subscribe(model: SubscriptionModelLiteral, event: SubscriptionEventLiteral, filters: Optional[SubscriptionFilters] = None) → pnwkit.new.Subscription[Any]
Create a new subscription with this QueryKit.
- Parameters
model (SubscriptionModelLiteral) – The model to receive events about
event (SubscriptionEventLiteral) –
filters (Optional[SubscriptionFilters]) – The parameters to provide to the subscription to filter the events
- Returns
A Subscription that can be subscribed too.
- Return type
Subscription[Any]
Query¶
- class pnwkit.new.Query(kit: QueryKit, *fields: Field, variables: Optional[Dict[str, Variable]] = None, variable_values: Optional[Dict[str, Any]] = None, hash: Optional[str] = None)¶
- clone() → Self¶
Clone a Query
- Returns
Returns a cloned instance of the query
- Return type
Self
- get(headers: Optional[Dict[str, Any]] = None) → R¶
Fetch the results of the query synchronously using requests
- Parameters
headers (Optional[Dict[str, Any]], optional) – Any additional headers to pass with the query, by default None
- Returns
The
Result
of the Query- Return type
R
- async get_async(headers: Optional[Dict[str, Any]] = None) → R¶
Fetch the results of the query asynchronously using aiohttp, simply using the
await
statement on the Query will also call this method- Parameters
headers (Optional[Dict[str, Any]], optional) – Any additional headers to pass with the query, by default None
- Returns
The
Result
of the Query- Return type
R
- paginate(field: Literal[nations]) → pnwkit.new.Paginator[pnwkit.data.Nation]¶
- paginate(field: Literal[alliances]) → pnwkit.new.Paginator[pnwkit.data.Alliance]
- paginate(field: Literal[tradeprices]) → pnwkit.new.Paginator[pnwkit.data.Tradeprice]
- paginate(field: Literal[trades]) → pnwkit.new.Paginator[pnwkit.data.Trade]
- paginate(field: Literal[wars]) → pnwkit.new.Paginator[pnwkit.data.War]
- paginate(field: Literal[bounties]) → pnwkit.new.Paginator[pnwkit.data.Bounty]
- paginate(field: Literal[warattacks]) → pnwkit.new.Paginator[pnwkit.data.WarAttack]
- paginate(field: Literal[treaties]) → pnwkit.new.Paginator[pnwkit.data.Treaty]
- paginate(field: Literal[cities]) → pnwkit.new.Paginator[pnwkit.data.City]
- paginate(field: Literal[bankrecs]) → pnwkit.new.Paginator[pnwkit.data.Bankrec]
- paginate(field: Literal[baseball_games]) → pnwkit.new.Paginator[pnwkit.data.BBGame]
- paginate(field: Literal[baseball_teams]) → pnwkit.new.Paginator[pnwkit.data.BBTeam]
- paginate(field: Literal[baseball_players]) → pnwkit.new.Paginator[pnwkit.data.BBPlayer]
- paginate(field: Literal[treasure_trades]) → pnwkit.new.Paginator[pnwkit.data.TreasureTrade]
- paginate(field: Literal[embargoes]) → pnwkit.new.Paginator[pnwkit.data.Embargo]
- paginate(field: str) → pnwkit.new.Paginator[Any]
Get a
Paginator
for paginating through a specific field- Parameters
field (str) – The field to paginate
- Returns
The Paginator with it’s type corresponding to the type of the field provided
- Return type
Paginator[Any]
- query(field: RootFieldLiteral, arguments: Dict[str, Union[Argument, Sequence[Argument]]], *fields: FieldValue) → Self¶
Query another field with this Query
- fieldRootFieldLiteral
The name of the root field to query
- argumentsDict[str, Union[Argument, Sequence[Argument]]]
The parameters to provide to the field to filter results
- fields: Union[str,
Field
] The fields to fetch in the query
- variables: MutableMapping[str, Any]
The values of any variables specified in the query
- Returns
Returns the Query for support for method chaining
- Return type
Self
- query_as(field: RootFieldLiteral, alias: str, arguments: Dict[str, Union[Argument, Sequence[Argument]]], *fields: FieldValue) → Self¶
Query another field with this Query
- Parameters
field (RootFieldLiteral) – The name of the root field to query
alias (
str
) – The name to have to results of this query returned by on the Resultarguments (Dict[
str
, Union[Argument, Sequence[Argument]]]) – The parameters to provide to the field to filter resultsfields (Union[str,
Field
]) – The fields to fetch in the queryvariables (MutableMapping[
str
, Any]) – The values of any variables specified in the query
- Returns
Returns the Query for support for method chaining
- Return type
Self
- set_variables(**variables: Any) → Self¶
Set variable values on the query
- variablesAny
The values to set for individual variables
- Returns
The Query for support with method chaining
- Return type
Self
Field¶
Paginator¶
- class pnwkit.new.Paginator(kit: pnwkit.new.QueryKit, query: pnwkit.new.Query[pnwkit.new.Result])¶
Represents a Paginator for use in fetching paginated values, designed for use in a
for
/async for
loop- batch(size: int, /) → Self¶
Batch the queries used to fill the paginator, will run multiple queries simultaneously corresponding to the size provided, only works when using asynchronous iteration
- Parameters
size (int) – The size of each batch of requests
- Returns
Returns the Paginator for use in method chaining
- Return type
Self
Mutation¶
- class pnwkit.new.Mutation(kit: QueryKit, *fields: Field, variables: Optional[Dict[str, Variable]] = None, variable_values: Optional[Dict[str, Any]] = None, hash: Optional[str] = None)¶
Supports all methods of
Query
where applicable- get(headers: Optional[Dict[str, Any]] = None) → R¶
Fetch the results of the query synchronously using requests
- Parameters
headers (Optional[Dict[str, Any]], optional) – Any additional headers to pass with the query, by default None
- Returns
The
Result
of the Query- Return type
R
- async get_async(headers: Optional[Dict[str, Any]] = None) → R¶
Fetch the results of the query asynchronously using aiohttp, simply using the
await
statement on the Query will also call this method- Parameters
headers (Optional[Dict[str, Any]], optional) – Any additional headers to pass with the query, by default None
- Returns
The
Result
of the Query- Return type
R
Subscription¶
- class pnwkit.new.Subscription(kit: QueryKit, model: SubscriptionModelLiteral, event: SubscriptionEventLiteral, filters: SubscriptionFilters, channel: Optional[str] = None, callbacks: Optional[List[Callback[T]]] = None)¶
- async classmethod subscribe(kit: QueryKit, model: SubscriptionModelLiteral, event: SubscriptionEventLiteral, filters: SubscriptionFilters, *callbacks: Callback[T]) → Self¶
Subscribe to the subscription, events can be received through asynchronous iteration (an
async for
loop) or through the provided callbacks- Parameters
callbacks (Callback[T]) – A list of async functions to call when an event is received
- async unsubscribe() → None¶
Unsubscribe from the subscription