Python

You can run multiple async tasks simultaneously in Python using the asyncio native library. This allows you to create and manage asynchronous tasks, such as coroutines, and run them concurrently. Let’s have the following async method, that counts to the length of the given name and returns the name. import asyncio from typing import Awaitable async def count_word(name: str) -> Awaitable[str]: if len(name) > 8: raise ValueError(f"{name} is too long...") for ii in range(len(name)): print(name, ii) await asyncio.sleep(1) return name Now, running this task twice: await count_word("first"), await count_word("second") first 0 first 1 first 2 first 3 first 4 second 0 second 1 second 2 second 3 second 4 second 5 ('first', 'second') Will not run the two tasks concurrently because it’s using await, which means that it will wait for the first task to complete before starting the second task. If you want to run these tasks concurrently, you should use asyncio.gather(): ...

ChatPDF providers, where you can question large files with large language models (LLM’s), are sprouting like mushrooms. The technique is mainly based on vector embedding with a vector index or vector database; based on the question, semantically relevant chunks from the file are provided to the LLM so it can compose an answer. While this technique is cool, it is limited when you ask a question that spans the entire text, such as generating a summary, since that requires not a couple of chunks, but the full text. ...

Had a complete headache trying to figure out how a decorator as a class can maintain the possible async properties of a method. The solution is actually very simple. When called, use inspect.iscoroutinefunction to check whether it is a coroutine, and return again an async method! The example adds given paths to a registry, import inspect from functools import wraps paths_registry = [] class route(object): def __init__(self, path: str, **kwargs) -> None: self._path = path def __call__(self, fn): paths_registry.append( { "path": self._path, } ) @wraps(fn) def decorated(*args, **kwargs): return fn(*args, **kwargs) @wraps(fn) async def asyncdecorated(*args, **kwargs): return await fn(*args, **kwargs) if inspect.iscoroutinefunction(fn): return asyncdecorated return decorated A method with this decorator would look like: ...

This is a simple demonstration of a domain event factory in Python. I assume you are familiar with the Factory Method Pattern. I also use the pydantic package for attribute validation. When implemented, we can use the factory to create immutable domain events with a homogenous data structure across instances of the same type. The metadata is generated by the underlying BaseEvent. In this approach we always produces complete events. Start with all our imports: ...

While reading into implementing DDD, there is often a plea for the use of immutable objects. The main motivation is that an object that is initially valid always remains valid; there is no later verification or validation required. Secondly, working with an immutable object cannot cause any side effects. Some data objects in Python are immutable, the dataclasses themselve are not. Let’s have this simple class: class SimpleClass: def __init__(self, attr1: int): self.attr1 = attr1 With this setup, you can change any of its attributes making the object not immutable: ...