This module provides utilities for common tasks involving the with statement.
contextlib.closing(thing)
Return a context manager that closes thing upon completion of the block. This is basically equivalent to:
from contextlib import contextmanager@contextmanagerdef closing(thing): try: yield thing finally: thing.close()
from contextlib import closingfrom urllib.request import urlopenwith closing(urlopen('https://www.python.org')) as page: for line in page: print(line)
Note
Most types managing resources support the context manager protocol, which closes thing on leaving the with statement.
As such, closing() is most useful for third party types that don’t support context managers.
This example is purely for illustration purposes, as urlopen() would normally be used in a context manager.
contextlib.suppress(*exceptions)
Return a context manager that suppresses any of the specified exceptions if they occur in the body of a with statement
and then resumes execution with the first statement following the end of the with statement.
from contextlib import suppresswith suppress(FileNotFoundError): os.remove('somefile.tmp')with suppress(FileNotFoundError): os.remove('someotherfile.tmp')
🧠 Changed in version 3.9: Class methods can now wrap other descriptors
such as property().
Note
🦖 Changed in version 3.10: Class methods now inherit the method attributes
(__module__, __name__, __qualname__,__doc__ and __annotations__) and have a new __wrapped__ attribute.
Note
🔥 Changed in version 3.11: Class methods can no longer wrap other descriptors
such as property().
Generators
Note
Have you ever had to work with a dataset so large that it overwhelmed your machine’s memory?
Or maybe you have a complex function that needs to maintain an internal state every time it’s called, but the function is too small to justify creating its own class. In these cases and more, generators and the Python yield statement are here to help.
Introduced with PEP 255, generator functions are a special kind of function that return a lazy iterator. These are objects that you can loop over like a list. However, unlike lists, lazy iterators do not store their contents in memory.
A generator expression yields a new generator object. Its syntax is the same as for comprehensions, except
that it is enclosed in parentheses instead of brackets or curly braces.
Note
In this case you lose the advantage of generators, but there are situations where that might be useful.
temperatures = list(temperatures)
Operator
The operator
module exports a set of efficient functions corresponding to the intrinsic operators of Python.
For example, operator.add(x, y) is equivalent to the expression x+y.
Bisect
This module provides support for maintaining a list in sorted order without having to sort the list after each insertion. For long lists of items with expensive comparison operations, this can be an improvement over linear searches or frequent resorting.
Locate the insertion point for x in a to maintain sorted order.
The parameters lo and hi may be used to specify a subset of the list which should be considered;
by default the entire list is used.
If x is already present in a, the insertion point will be before (to the left of) any existing entries.
The return value is suitable for use as the first parameter to list.insert() assuming that a is already sorted.
The returned insertion point ip partitions the array a into two slices such that
all(elem < x for elem in a[lo : ip]) is true for the left slice and all(elem >= x for elem in a[ip : hi]) is true for the right slice.
key specifies a key function of one argument that is used to extract
a comparison key from each element in the array. To support searching complex records, the key function is not applied to the x value.
If key is None, the elements are compared directly and no key function is called.
Similar to bisect_left(),
but returns an insertion point which comes after (to the right of) any existing entries of x in a.
The returned insertion point ip partitions the array a into two slices such that
all(elem <= x for elem in a[lo : ip]) is true for the left slice and
all(elem > x for elem in a[ip : hi]) is true for the right slice.
This module implements a number of iterator
building blocks inspired by constructs from APL, Haskell, and SML.
Each has been recast in a form suitable for Python.
The module standardizes a core set of fast, memory efficient tools that are useful by themselves or in combination. Together, they form an
“iterator algebra” making it possible to construct specialized tools succinctly and efficiently in pure Python.
r-length tuples, in sorted order, with repeated elements
Batched
itertools.batched(iterable, n)
Batch data from the iterable into tuples of length n.
The last batch may be shorter than n.
Loops over the input iterable and accumulates data into tuples up to size n.
The input is consumed lazily, just enough to fill a batch.
The result is yielded as soon as the batch is full or when the input iterable is exhausted:
Python’s iterators and iterables are two different but related tools that come in handy when
you need to iterate over a data stream or container.
Iterators power and control the iteration process,
while iterables typically hold data that you want to iterate over one value at a time.
Formatted Strings
A formatted string literal or f-string is a string literal that is prefixed with
f or F. These strings may contain replacement fields,
which are expressions delimited by curly braces {}.
While other string literals always have a constant value, formatted strings are really expressions evaluated at run time.
The parts of the string outside curly braces are treated literally, except that any doubled curly braces {{ or }}
are replaced with the corresponding single curly brace. A single opening curly bracket { marks a replacement field,
which starts with a Python expression.
To display both the expression text and its value after evaluation, (useful in debugging), an equal sign =
may be added after the expression.
A conversion field, introduced by an exclamation point ! may follow.
A format specifier may also be appended, introduced by a colon :.
A replacement field ends with a closing curly bracket }.
If a conversion is specified, the result of evaluating the expression is converted before formatting.
Conversion !s calls str() on the result,
!r calls repr(),
and !a calls ascii().
The result is then formatted using the format() protocol.
The format specifier is passed to the __format__()
method of the expression or conversion result. An empty string is passed when the format specifier is omitted.
The formatted result is then included in the final value of the whole string.
:::note
🔧 Changed in version 3.12: Prior to Python 3.12, reuse of the same quoting type of the outer f-string inside a replacement field was not possible.
:::
Backslashes are also allowed in replacement fields and are evaluated the same way as in any other context:
a = ["a", "b", "c"]print(f"List a contains:\n{"\n".join(a)}")
:::note
🔧 Changed in version 3.12: Prior to Python 3.12, backslashes were not permitted inside an f-string replacement field.
:::
Configuration 🔧
I love the way we can configure Golang applications WITH TYPES, but this library is in Python, and seems nice to me for doing the same in Python.
Sometimes I don’t want to set up the giant Django so, I have these options.
Sanic and FastAPI both are asynchronous and says they have good performance.
Sanic
Simple and lightweight: Intuitive API with smart defaults and no bloat allows you to get straight to work building your app.
Unopinionated and flexible: Build the way you want to build without letting your tooling constrain you.
Performant and scalable: Built from the ground up with speed and scalability as a main concern. It is ready to power web applications big and small.
Production ready: Out of the box, it comes bundled with a web server ready to power your web applications.
Trusted by millions: Sanic is one of the overall most popular frameworks on PyPI, and the top async enabled framework
Community driven: The project is maintained and run by the community for the community.
After installing, Sanic has all the tools you need for a scalable, production-grade server—out of the box!
Running Sanic with TLS enabled is as simple as passing it the file paths…
Up and running with web sockets in no time using the websockets package.
Serving static files is of course intuitive and easy. Just name an endpoint and either a file or directory that should be served.
Beginning or ending a route with functionality is as simple as adding a decorator.
Raising errors will intuitively result in proper HTTP errors:
Check in on your live, running applications (whether local or remote).
In addition to the tools that Sanic comes with, the officially supported Sanic Extensions provides lots of extra goodies to make development easier.
Dependency injection is a method to add arguments to a route handler based upon the defined function signature. Specifically,
it looks at the type annotations of the arguments in the handler. This can be useful in a number of cases like:
Within the Python programming language, a diverse set of HTTP client libraries exists to facilitate communication
with web servers. These libraries enable the programmatic construction and transmission of HTTP requests.
The urllib module, a built-in library, offers a foundational approach to HTTP communication.
The requests library, known for its ease of use and rich feature set,
is a widely adopted choice.
The aiohttp library caters to projects requiring asynchronous operations,
ideal for non-blocking communication patterns.
Writing REST API in a Django application using Django REST Framework is awesome, following links are the important concepts of the DRF (Django REST Framework):
Django’s database layer provides various ways to help developers get the most out of their databases.
As general programming practice, this goes without saying. Find out what queries you are doing and what they are
costing you. Use QuerySet.explain() to understand how specific QuerySets are executed by your database.
To avoid performance problems, it is important to understand:
That QuerySets are lazy.
When they are evaluated.
How the data is held in memory.
Use iterator(),
When you have a lot of objects, the caching behavior of the QuerySet can cause a large amount
of memory to be used. In this case, iterator() may help.
Dataclasses
Using data-classes to define request and response in Django REST Framework. There are cases in which your request or
response is not a model, in those cases you can define them as a dataclass using the following library.
Having reusable filters for models in Django REST Framework with Django-filter. These filters help you to write
viewsets easier and give client developers vast choices in getting the data.
Typer, build great CLIs. Easy to code. Based on Python type hints.
Create a typer.Typer() app, and create two sub commands with their parameters
import typerapp = typer.Typer()@app.command()def hello(name: str): print(f"Hello {name}")@app.command()def goodbye(name: str, formal: bool = False): if formal: print(f"Goodbye Ms. {name}. Have a good day.") else: print(f"Bye {name}!")if __name__ == "__main__": app()
python main.py --help Usage: main.py [OPTIONS] COMMAND [ARGS]...╭─ Options ─────────────────────────────────────────╮│ --install-completion Install completion ││ for the current ││ shell. ││ --show-completion Show completion for ││ the current shell, ││ to copy it or ││ customize the ││ installation. ││ --help Show this message ││ and exit. │╰───────────────────────────────────────────────────╯╭─ Commands ────────────────────────────────────────╮│ goodbye ││ hello │╰───────────────────────────────────────────────────╯
python main.py hello --help Usage: main.py hello [OPTIONS] NAME╭─ Arguments ───────────────────────────────────────╮│ * name TEXT [default: None] [required] │╰───────────────────────────────────────────────────╯╭─ Options ─────────────────────────────────────────╮│ --help Show this message and exit. │╰───────────────────────────────────────────────────╯
You can also display beautiful and more complex information using Rich.
Using it for reading/writing CSV is a better way than any other console application because you can actually do the query things after the reading phase.
The function below takes and returns a string and is annotated as follows:
def greeting(name: str) -> str: return 'Hello ' + name
In the function greeting, the argument name is expected to be of type str
and the return type str. Subtypes are accepted as arguments.
Type aliases
A type alias is defined using the type statement,
which creates an instance of TypeAliasType.
In this example, Vector and list[float] will be treated equivalently by static type checkers:
type Vector = list[float]def scale(scalar: float, vector: Vector) -> Vector: return [scalar * num for num in vector]# passes type checking; a list of floats qualifies as a Vector.new_vector = scale(2.0, [1.0, -4.2, 5.4])
Type aliases are useful for simplifying complex type signatures.
The subscription syntax must always be used with exactly two values: the argument list and the return type.
The argument list must be a list of types, a ParamSpec,
Concatenate, or an ellipsis.
The return type must be a single type.
If a literal ellipsis ... is given as the argument list, it indicates that a callable with any arbitrary parameter list would be acceptable:
def concat(x: str, y: str) -> str:return x + yx: Callable[..., str]x = str# OKx = concat# Also OK
:::note
⚠️ Callable cannot express complex signatures such as functions that take a variadic number of arguments,
overloaded functions, or functions that have keyword-only parameters.
However, these signatures can be expressed by defining a Protocol
class with a __call__() method.
:::
from collections.abc import Iterablefrom typing import Protocolclass Combiner(Protocol): def __call__(self, *vals: bytes, maxlen: int | None = None) -> list[bytes]: ...def batch_proc(data: Iterable[bytes], cb_results: Combiner) -> bytes: for item in data: ...def good_cb(*vals: bytes, maxlen: int | None = None) -> list[bytes]: ...def bad_cb(*vals: bytes, maxitems: int | None) -> list[bytes]: ...batch_proc([], good_cb) # OKbatch_proc([], bad_cb) # Error! Argument 2 has incompatible type because of # different name and kind in the callback
Generics
Since type information about objects kept in containers cannot be statically inferred in a generic way, many container classes in the standard library support subscription to denote the expected types of container elements.
from collections.abc import Mapping, Sequenceclass Employee: ...# Sequence[Employee] indicates that all elements in the sequence# must be instances of "Employee".# Mapping[str, str] indicates that all keys and all values in the mapping# must be strings.def notify_by_email(employees: Sequence[Employee], overrides: Mapping[str, str]) -> None: ...
from collections.abc import Sequencedef first[T](l: Sequence[T]) -> T: # Function is generic over the TypeVar "T" return l[0]
:::note
⚠️ Changed in version 3.12: Syntactic support for generics is new in Python 3.12.
:::
For most containers in Python, the typing system assumes that all elements in the container will be of the same type.
list only accepts one type argument.
Mapping only accepts two type arguments:
the first indicates the type of the keys, and the second indicates the type of the values.
from collections.abc import Mapping# Type checker will infer that all elements in ``x`` are meant to be intsx: list[int] = []# Type checker error: ``list`` only accepts a single type argument:y: list[int, str] = [1, 'foo']# Type checker will infer that all keys in ``z`` are meant to be strings,# and that all values in ``z`` are meant to be either strings or intsz: Mapping[str, str | int] = {}
Unlike most other Python containers, however, it is common in idiomatic Python code for tuples to have elements which are not all the same type.
For this reason, tuples are special-cased in Python’s typing system. tuple
accepts any number of type arguments:
# OK: ``x`` is assigned to a tuple of length 1 where the sole element is an intx: tuple[int] = (5,)# OK: ``y`` is assigned to a tuple of length 2;# element 1 is an int, element 2 is a stry: tuple[int, str] = (5, "foo")# Error: the type annotation indicates a tuple of length 1,# but ``z`` has been assigned to a tuple of length 3z: tuple[int] = (1, 2, 3)
To denote a tuple which could be of any length, and in which all elements are of the same type T, use tuple[T, ...].
To denote an empty tuple, use tuple[()]. Using plain tuple as an annotation is equivalent to using tuple[Any, ...].
The type of class objects
A variable annotated with C may accept a value of type C.
In contrast, a variable annotated with type[C] (or typing.Type[C])
may accept values that are classes themselves - specifically, it will accept the class object of C.
a = 3 # Has type ``int``b = int # Has type ``type[int]``c = type(a) # Also has type ``type[int]``
Note that type[C] is covariant:
:::note
🧠 Covariance and contravariance are terms that refer to the ability to use a more derived type (more specific) or
a less derived type (less specific) than originally specified
:::
class User: ...class ProUser(User): ...class TeamUser(User): ...def make_new_user(user_class: type[User]) -> User: # ... return user_class()make_new_user(User) # OKmake_new_user(ProUser) # Also OK: ``type[ProUser]`` is a subtype of ``type[User]``make_new_user(TeamUser) # Still finemake_new_user(User()) # Error: expected ``type[User]`` but got ``User``make_new_user(int) # Error: ``type[int]`` is not a subtype of ``type[User]``
The only legal parameters for type are classes,
Any,
type variables, and unions of any of these types.
One of the primary ways of defining schema in Pydantic is via models.
Models are simply classes which inherit from
pydantic.BaseModel
and define fields as annotated attributes.
You can think of models as similar to structs in languages like C, or
as the requirements of a single endpoint in an API.
Models share many similarities with Python’s dataclasses,
but have been designed with some subtle-yet-important differences that streamline
certain workflows related to validation, serialization, and JSON schema generation.
Untrusted data can be passed to a model and, after parsing and validation,
Pydantic guarantees that the fields of the resultant model instance
will conform to the field types defined on the model.
Beyond accessing model attributes directly via their field names
(e.g. model.foobar), models can be converted, dumped, serialized,
and exported in a number of ways.
The Field
function is used to customize and add metadata to fields of models.
Examples
class Comparable(pydantic.BaseModel): """ ... It only handles deserialization from the API response. """ class Config: use_enum_values = True @pydantic.field_validator("sold_date", mode="before") @classmethod def sold_date_from_rfc1123(cls, v: str): """ sold date is in rfc1123, date-only, etc. formats and pydantic cannot parse it. """ for fmt in [ "%a, %d %b %Y %H:%M:%S GMT", "%Y-%m-%d", ]: try: return datetime.datetime.strptime(v, fmt).date() except ValueError: continue raise ValueError(f"unexpected date format for 'sold_date' {v}") address: str = pydantic.Field(validation_alias="Address") unit_number: int | typing.Literal["NA"] = pydantic.Field(validation_alias="AptUnit") sold_date: datetime.date = pydantic.Field(validation_alias="SoldDate") sold_price: float = pydantic.Field(validation_alias="SoldPrice")
Asyncio is used as a foundation for multiple Python asynchronous frameworks that provide high-performance
network and web-servers, database connection libraries, distributed task queues, etc.
Running an asyncio Program
Runners are built on top of an event loop
with the aim to simplify async code usage for common wide-spread scenarios.
This function runs the passed coroutine, taking care of managing the asyncio event loop,
finalizing asynchronous generators, and closing the executor.
:::note
This function cannot be called when another asyncio event loop is running in the same thread.
:::
If debug is True, the event loop will be run in debug mode.
False disables debug mode explicitly.
None is used to respect the global
Debug Mode settings.
If loop_factory is not None, it is used to create a new event loop;
otherwise asyncio.new_event_loop()
is used.
The loop is closed at the end. This function should be used as a main entry point for asyncio programs,
and should ideally only be called once.
It is recommended to use loop_factory to configure the event loop instead of policies.
If debug is True, the event loop will be run in debug mode.
False disables debug mode explicitly.
None is used to respect the global Debug Mode settings.
loop_factory could be used for overriding the loop creation.
It is the responsibility of the loop_factory to set the created loop as the current one.
By default, asyncio.new_event_loop()
is used and set as current event loop with asyncio.set_event_loop()
if loop_factory is None.
Basically, asyncio.run()
example can be rewritten with the runner usage:
async def main(): await asyncio.sleep(1) print('hello')with asyncio.Runner() as runner: runner.run(main())
run(coro, *, context=None)
Run a coroutinecoro in the embedded loop.
Return the coroutine’s result or raise its
exception. An optional keyword-only context argument allows specifying a custom
contextvars.Context
for the coro to run in. The runner’s default context is used if None.
This function cannot be called when another asyncio event loop is running in the same thread.
close()
Close the runner. Finalize asynchronous generators, shutdown default executor,
close the event loop and release embedded contextvars.Context.
get_loop()
Return the event loop associated with the runner instance.
Coroutines
Coroutines
declared with the async/await syntax is the preferred way of writing asyncio applications.
To actually run a coroutine, asyncio provides the following mechanisms:
The asyncio.run() function to run the top-level entry point “main()” function.
Awaiting on a coroutine. The following snippet of code will print “hello” after waiting for 1 second, and then print “world” after waiting for another 2 seconds:
async def main(): task1 = asyncio.create_task( say_after(1, 'hello')) task2 = asyncio.create_task( say_after(2, 'world')) print(f"started at {time.strftime('%X')}") # Wait until both tasks are completed (should take # around 2 seconds.) # They ran with create_task and here we only wait # for their results. await task1 await task2 print(f"finished at {time.strftime('%X')}")
The asyncio.TaskGroup class provides a more modern alternative to
create_task(). Using this API, the last example becomes:
async def main(): async with asyncio.TaskGroup() as tg: task1 = tg.create_task( say_after(1, 'hello')) task2 = tg.create_task( say_after(2, 'world')) print(f"started at {time.strftime('%X')}") # The await is implicit when the context manager exits. print(f"finished at {time.strftime('%X')}")
Awaitables
We say that an object is an awaitable object if it can be used in
an await expression.
Many asyncio APIs are designed to accept awaitables.
There are three main types of awaitable objects:
coroutines
Tasks
Futures
:::note
⚠️ A coroutine function: an async def function.
A coroutine object: an object returned by calling a coroutine function.
:::
:::note
🔥 Tasks are used to schedule coroutines concurrently.
:::
:::note
🚧 A new alternative to create and run tasks concurrently and wait for their completion is
asyncio.TaskGroup`.
TaskGroup provides stronger safety guarantees than gather for scheduling a nesting of subtasks:
if a task (or a subtask, a task scheduled by a task) raises an exception,
TaskGroup will, while gather will not, cancel the remaining scheduled tasks.
:::
When using this factory (via loop.set_task_factory(asyncio.eager_task_factory)),
coroutines begin execution synchronously during Task construction.
Tasks are only scheduled on the event loop if they block.
This can be a performance improvement as the overhead of
loop scheduling is avoided for coroutines that complete synchronously.
A common example where this is beneficial is coroutines which employ caching or memoization
to avoid actual I/O when possible.
:::note
🐼 Immediate execution of the coroutine is a semantic change.
If the coroutine returns or raises, the task is never scheduled to the event loop.
If the coroutine execution blocks, the task is scheduled to the event loop.
This change may introduce behavior changes to existing applications.
For example, the application’s task execution order is likely to change.
:::
AIOfiles
aiofiles is an Apache2 licensed library,
written in Python, for handling local disk files in asyncio applications.
:::note
Don’t create a session per request. Most likely you need a session per application which performs all requests together.
More complex cases may require a session per site, e.g. one for GitHub and other one for Facebook APIs. Anyway making a session for every request is a very bad idea.
A session contains a connection pool inside. Connection reuse and keep-alive (both are on by default) may speed up total performance.
:::
A session context manager usage is not mandatory but await session.close() method should be called in this case.
Streaming Response Content
While methods read(),
json()
and text()
are very convenient you should use them carefully. All these methods load the whole response in memory.
For example if you want to download several gigabyte sized files, these methods will load all the data in memory.
Instead, you can use the content attribute.
It is an instance of the aiohttp.StreamReader class.
The gzip and deflate transfer-encodings are automatically decoded for you:
async with session.get('https://api.github.com/events') as resp: await resp.content.read(10)
In general, however, you should use a pattern like this to save what is being streamed to a file:
with open(filename, *'wb'*) as fd: async for chunk in resp.content.iter_chunked(chunk_size): fd.write(chunk)
/* GC information is stored BEFORE the object structure. */typedef union _gc_head{ struct { union _gc_head *gc_next; union _gc_head *gc_prev; Py_ssize_t gc_refs; } gc; long double dummy; /* force worst-case alignment */} PyGC_Head;
The theory was that every time we did a collection, it would update the gc_refs with ob_refcnt for all tracked objects
— but unfortunately this write-operation, caused memory pages to be COW-ed. A next obvious solution was to move all the
head to another chunk of memory and store densely.