Quickstart

Learn the most common AlexBerUtils patterns with practical, working code examples.

This page shows four practical examples that cover the most commonly used parts of AlexBerUtils. Each example uses real code patterns from the library.

Install the package before running any examples: python -m pip install -U alex-ber-utils For YAML examples, also run: python -m pip install alex-ber-utils[yml]

Thread-safe object access with LockingProxy

LockingProxy wraps any object and ensures every attribute access, method call, item get/set, and iteration is guarded by a lock. It composes multiple locking mixins (LockingAccessMixin, LockingIterableMixin, LockingCallableMixin, etc.) into a single transparent proxy.

1. Import the proxy and lock

   from threading import RLock
   from alexber.utils.thread_locals import LockingProxy
   
2. Wrap a shared object

   Create an RLock, wrap your object with LockingProxy, and use the proxy from any thread:

   import threading
   
   shared_data = {"count": 0}
   lock = threading.RLock()
   
   # Wrap the dict — all access through `proxy` is now guarded by `lock`
   proxy = LockingProxy(obj=shared_data, lock=lock)
   
   def worker():
       # __getitem__ and __setitem__ are both lock-protected
       proxy["count"] = proxy["count"] + 1
   
   threads = [threading.Thread(target=worker) for _ in range(10)]
   for t in threads:
       t.start()
   for t in threads:
       t.join()
   
   print(shared_data["count"])  # 10
   
3. Use it as a context manager

   LockingProxy also supports with and async with blocks for explicit critical sections:

   with proxy:
       # entire block is lock-protected
       proxy["count"] += 1
   

LockingProxy handles both sync and async methods automatically. If the wrapped object has async def methods, the proxy wraps them with async with self._lock instead.

---

Running async code from a sync context with lift_to_async

lift_to_async lets you call an async function from synchronous code while preserving the current ContextVar context. This is useful in worker threads that need to drive async operations without owning an event loop.

1. Initialize the event loop config

   Call initConfig() once during application startup from an async context (it requires a running event loop):

   import asyncio
   from alexber.utils.thread_locals import initConfig
   
   async def startup():
       # Must be called from a coroutine so a running loop is available
       initConfig()
   
   asyncio.run(startup())
   
2. Call an async function from sync code

   After initialization, any synchronous code — including code running in a worker thread — can invoke lift_to_async:

   from alexber.utils.thread_locals import lift_to_async
   
   async def fetch_data(url: str) -> str:
       # any async logic here
       await asyncio.sleep(0)  # simulate I/O
       return f"data from {url}"
   
   # Called from synchronous code in a non-main thread
   result = lift_to_async(fetch_data, "https://example.com/api")
   print(result)  # "data from https://example.com/api"
   

lift_to_async must be called from a non-main thread when an event loop is already running. If you call it from the main thread while the loop is active, it raises RuntimeError. Use asyncio.to_thread() to move synchronous work off the main thread first.

---

Parsing app configuration with parse_config

parse_config in init_app_conf merges YAML configuration files with CLI arguments. It supports profile layering (config.yml → config-dev.yml → config-local.yml) and implicit type conversion.

1. Install the yml extra

   python -m pip install alex-ber-utils[yml]
   
2. Create a config.yml file

   # config.yml
   general:
     profiles:
       - dev
   
   database:
     host: localhost
     port: 5432
     name: mydb
   
3. Initialize and parse

   Call initConfig() once before your first parse_config() call, then use the returned dict throughout your application:

   import alexber.utils.ymlparsers as ymlparsers
   import alexber.utils.init_app_conf as init_app_conf
   
   # Initialize ymlparsers (required before parse_config)
   ymlparsers.initConfig()
   
   # Initialize init_app_conf with default settings
   init_app_conf.initConfig()
   
   # Parse: merges config.yml (and any profile overrides) with CLI args
   config = init_app_conf.parse_config()
   
   db_host = config["database"]["host"]   # "localhost" (or CLI override)
   db_port = config["database"]["port"]   # 5432 (int, not string)
   print(db_host, db_port)
   
4. Override with CLI arguments

   Any key in the YAML file can be overridden at runtime by passing --key=value on the command line:

   python myapp.py --database.host=prod-db.internal --database.port=5433
   

   parse_config automatically converts "5433" to the integer 5433 (implicit type conversion is on by default).

The general.profiles key controls which profile YAML files are layered on top of the base config.yml. For example, a profile of ["dev", "local"] causes config-dev.yml then config-local.yml to be merged in order, with later files taking precedence.

---

Simple YAML loading with ymlparsers.load

ymlparsers.load is a thin wrapper around HiYaPyCo that adds Jinja2 variable substitution and returns an OrderedDict. Use it when you need low-level YAML loading without the full parse_config pipeline.

1. Install the yml extra

   python -m pip install alex-ber-utils[yml]
   
2. Initialize and load a YAML file

   import alexber.utils.ymlparsers as ymlparsers
   
   # Initialize with default settings (must be called once)
   ymlparsers.initConfig()
   
   # Load a single YAML file
   data = ymlparsers.load("config.yml")
   print(data["database"]["host"])  # "localhost"
   
3. Load multiple YAML files (merge)

   Pass multiple file paths to merge them in order. Later files override earlier ones:

   data = ymlparsers.load("config.yml", "config-dev.yml")
   
4. Disable variable substitution

   Use the DisableVarSubst context manager when you want raw YAML values without Jinja2 rendering:

   with ymlparsers.DisableVarSubst():
       data = ymlparsers.load("config.yml")
   

ymlparsers.load is the low-level API. For most applications, prefer init_app_conf.parse_config, which handles profile layering and CLI argument merging on top of ymlparsers.load.

---

Next steps

Concurrency & Async

Deep dive into RLock, LockingProxy, exec_in_executor, and AsyncExecutionQueue.

App configuration

Full reference for parse_config profile layering, white-listing, and type conversion.

YAML utilities

All options for ymlparsers.load, as_str, and Jinja2 variable substitution.

Async cache

LRU/LFU caching with TTL support for async workloads.