PyUtils

Python Utils

This package contains dependency-free Python utility functions used throughout the codebase.

Each utility should belong in its own file and be the default export.

These functions are not part of the module interface and are subject to change.

graphql.pyutils.camel_to_snake(s: str) → str: Convert from CamelCase to snake_case

graphql.pyutils.snake_to_camel(s: str, upper: bool = True) → str

Convert from snake_case to CamelCase

If upper is set, then convert to upper CamelCase, otherwise the first character keeps its case.

graphql.pyutils.cached_property(func)

graphql.pyutils.register_description(base: type) → None: Register a class that shall be accepted as a description.

graphql.pyutils.unregister_description(base: type) → None: Unregister a class that shall no more be accepted as a description.

graphql.pyutils.did_you_mean(suggestions: Sequence[str], sub_message: str | None = None) → str: Given [ A, B, C ] return ‘ Did you mean A, B, or C?’

graphql.pyutils.identity_func(x: T = Undefined, *_args: Any) → T: Return the first received argument.

graphql.pyutils.inspect(value: Any) → str

Inspect value and a return string representation for error messages.

Used to print values in error messages. We do not use repr() in order to not leak too much of the inner Python representation of unknown objects, and we do not use json.dumps() because not all objects can be serialized as JSON and we want to output strings with single quotes like Python repr() does it.

We also restrict the size of the representation by truncating strings and collections and allowing only a maximum recursion depth.

graphql.pyutils.is_awaitable(value: Any) → TypeGuard[Awaitable]

Return True if object can be passed to an await expression.

Instead of testing whether the object is an instance of abc.Awaitable, we check the existence of an __await__ attribute. This is much faster.

graphql.pyutils.is_async_iterable(value: Any) → TypeGuard[AsyncIterable]

Return True if object is an asynchronous iterable.

Instead of testing whether the object is an instance of abc.AsyncIterable, we check the existence of an __aiter__ attribute. This is much faster.

graphql.pyutils.is_collection(value: Any) → TypeGuard[Collection]: Check if value is a collection, but not a string or a mapping.

graphql.pyutils.is_iterable(value: Any) → TypeGuard[Iterable]: Check if value is an iterable, but not a string or a mapping.

graphql.pyutils.natural_comparison_key(key: str) → tuple

Comparison key function for sorting strings by natural sort order.

See: https://en.wikipedia.org/wiki/Natural_sort_order

class graphql.pyutils.AbortController

Bases: object

A controller object for aborting the execution of a GraphQL operation.

This mirrors the JavaScript AbortController Web API. Pass its signal as the abort_signal argument to execute (and related functions) and call abort() to cancel the execution.

abort(reason: Any = None) → None

Abort the operation, optionally specifying a reason.

The reason becomes the resulting GraphQL error: an exception is used as its original error, while any other value is wrapped by located_error as an “Unexpected error value”. If no reason is given, an AbortError with a generic message is used. Aborting more than once has no further effect.

signal: AbortSignal

class graphql.pyutils.AbortSignal

Bases: object

A signal object that communicates an abort request to the executor.

This mirrors the JavaScript AbortSignal Web API. The executor only ever inspects the synchronous aborted flag (and reason) at field boundaries, so any object exposing these two attributes can be used in place of this class. In addition to that pollable interface, this implementation is also awaitable via wait(), allowing resolvers to react to an abort of a running operation immediately instead of only at the next field boundary.

The signal is created and controlled through an AbortController.

aborted: bool

reason: Any

async wait() → Any: Wait until the signal is aborted and return the abort reason.

exception graphql.pyutils.AbortError

Bases: Exception

Error used as the default reason when an operation is aborted.

This is the Python counterpart of the AbortError DOMException that the JavaScript AbortController.abort() uses when called without a reason.

add_note(note, /): Add a note to the exception

args

with_traceback(tb, /): Set self.__traceback__ to tb and return self.

graphql.pyutils.AwaitableOrValue: alias of Awaitable[T] | T

class graphql.pyutils.BoxedAwaitableOrValue(value: T | Awaitable[T])

Bases: Generic[T]

Container for an Awaitable or a Value that updates itself.

A BoxedAwaitableOrValue is a container for a value or Awaitable where the value will be updated when the Awaitable has been awaited.

property pending_future: Future[T] | None: Get the still pending Future, or None if the value is already settled.

property value: T: Get the current value.

graphql.pyutils.suggestion_list(input_: str, options: Collection[str]) → list[str]

Get list with suggestions for a given input.

Given an invalid input string and list of valid options, returns a filtered list of valid options sorted based on their similarity with the input.

class graphql.pyutils.FrozenError

Bases: TypeError

Error when trying to change a frozen (read only) collection.

class graphql.pyutils.Path(prev: Path | None, key: str | int, typename: str | None)

Bases: NamedTuple

A generic path of string or integer indices

add_key(key: str | int, typename: str | None = None) → Path: Return a new Path containing the given key.

as_list() → list[str | int]: Return a list of the path keys.

count(value, /): Return number of occurrences of value.

index(value, start=0, stop=9223372036854775807, /)

Return first index of value.

Raises ValueError if the value is not present.

key: str | int: current index in the path (string or integer)

prev: Path | None: path with the previous indices

typename: str | None: name of the parent type to avoid path ambiguity

graphql.pyutils.print_path_list(path: Collection[str | int]) → str: Build a string describing the path.

class graphql.pyutils.SimplePubSub

Bases: object

A very simple publish-subscript system.

Creates an AsyncIterator from an EventEmitter.

Useful for mocking a PubSub system for tests.

emit(event: Any) → bool: Emit an event.

get_subscriber(transform: Callable | None = None) → SimplePubSubIterator: Return subscriber iterator

subscribers: set[Callable]

class graphql.pyutils.SimplePubSubIterator(pubsub: SimplePubSub, transform: Callable | None)

Bases: AsyncIterator

Async iterator used for subscriptions.

async aclose() → None: Close the iterator.

async empty_queue() → None: Empty the queue.

pull_queue: Queue[Future]

push_queue: Queue[Any]

async push_value(event: Any) → None: Push a new value.

graphql.pyutils.Undefined = Undefined

Symbol for undefined values

This singleton object is used to describe undefined or invalid values. It can be used in places where you would use undefined in GraphQL.js.

class graphql.pyutils.RefMap(items: Iterable[tuple[K, V]] | None = None)

Bases: MutableMapping[K, V]

A dictionary like object that allows mutable objects as keys.

This class keeps the insertion order like a normal dictionary.

Note that the implementation is limited to what is needed internally.

get(key: Any, default: Any = None) → Any: Get the mapped value for the given key.

items() → Iterator[tuple[K, V]]: Return an iterator over the key/value-pairs of the map.

keys() → Iterator[K]: Return an iterator over the keys of the map.

update(items: Iterable[tuple[K, V]] | None = None) → None: Update the map with the given key/value-pairs.

values() → Iterator[V]: Return an iterator over the values of the map.

class graphql.pyutils.RefSet(values: Iterable[T] | None = None)

Bases: MutableSet[T]

A set like object that allows mutable objects as elements.

This class keeps the insertion order unlike a normal set.

Note that the implementation is limited to what is needed internally.

add(value: T) → None: Add the given item to the set.

discard(value: T) → None: Remove the given item from the set if it exists.

remove(value: T) → None: Remove the given item from the set.

update(values: Iterable[T] | None = None) → None: Update the set with the given items.