universe-reads 0.1.5

Parallel UniVerse reader using uopy

universe-reads

Reads large volumes of records from a UniVerse DB using 5 concurrent sessions by default (configurable) via uopy.

Quick start

1) Install

python -m venv .venv
source .venv/bin/activate
pip install -e .

If uopy is installable via pip in your environment:

pip install -e ".[uopy]"

If uopy is not available on PyPI in your environment, install it using your internal method and keep the import name uopy.

For a quick run without installing the package, you can also do:

PYTHONPATH=src python -m universe_reads --dry-run 10000 --count-only

2) Configure

Set environment variables as you like for your own factory (optional).

universe-reads itself only needs:

• UV_FILE (optional if you pass --file)

You provide the UniVerse session creation via --session-factory.

Session factory (important)

This project uses multiprocessing with the spawn start method, which means worker processes must be able to import whatever they need.

Because of that, the recommended way to specify a session factory is a string import spec:

• "module.submodule:callable_name"

This is robust across processes and lets you reuse the same factory reference in many modules.

A) Minimal factory module

Example factory module my_uv.py:

import uopy


def make_session():
  return uopy.connect(host="...", account="...", user="...", password="...")

Then use it from the CLI:

UV_FILE=READS universe-reads --session-factory my_uv:make_session --count-only

Or programmatically:

from universe_reads import run


results = run(
  file_name="READS",
  session_factory="my_uv:make_session",
  sessions=5,
  count_only=True,
)

B) Why not pass a lambda/closure?

Avoid nested functions / lambdas for session_factory when using multiprocessing spawn.

Do NOT do:

from universe_reads import run


def make_factory():
  def factory():
    ...
  return factory


run(file_name="READS", session_factory=make_factory())

Prefer a top-level function in an importable module and pass it as a string spec.

C) Reuse across modules

If you’ll use this in a bunch of places, put the spec string in one location:

# settings.py
SESSION_FACTORY = "my_uv:make_session"

Then:

from settings import SESSION_FACTORY
from universe_reads import run


run(file_name="READS", session_factory=SESSION_FACTORY)

Or use the built-in env-based factory:

--session-factory universe_reads.session_factory:make_session

D) Pooling kwargs are optional (factory-friendly)

The CLI/library can pass pooling hints as keyword args:

• pooling_on
• min_pool_size
• max_pool_size

If your factory does not accept them, they are safely ignored.

Example of a kw-accepting factory:

import uopy


def make_session(*, pooling_on=None, min_pool_size=None, max_pool_size=None):
  kwargs = {"user": "...", "password": "..."}
  if pooling_on:
    kwargs["min_pool_size"] = min_pool_size or 1
    kwargs["max_pool_size"] = max_pool_size or 10
  return uopy.connect(**kwargs)

3) Run

Count-only (fastest):

UV_FILE=READS universe-reads \
  --session-factory my_uv:make_session \
  --count-only

Using the built-in env-based factory:

UV_HOST=... UV_ACCOUNT=... UV_USER=... UV_PASSWORD=... UV_FILE=READS \
  universe-reads --session-factory universe_reads.session_factory:make_session --count-only

More sessions (if you have licenses/CPU to spare):

UV_FILE=READS universe-reads \
  --session-factory my_uv:make_session \
  --sessions 10 \
  --count-only

Connection pooling (optional)

If your uopy environment has pooling enabled (often via uopy.ini), you can keep it optional by enabling it only in your session factory.

Important caveats:

• Pooling is per Python process. Since this project uses multiprocessing for parallel sessions, each worker process has its own pool.
• Pooling is not “multi-user” by itself. A pool is typically keyed by connection parameters (host/account/user/password). Different users generally mean separate pools.

With the built-in factory you can enable pooling hints via CLI flags:

universe-reads \
  --session-factory universe_reads.session_factory:make_session \
  --pooling \
  --min-pool-size 1 \
  --max-pool-size 10 \
  --count-only

Or via env vars:

export UV_POOLING_ON=1
export UV_MIN_POOL_SIZE=1
export UV_MAX_POOL_SIZE=10

universe-reads --session-factory universe_reads.session_factory:make_session --count-only

Write one NDJSON per worker:

UV_FILE=READS universe-reads \
  --session-factory my_uv:make_session \
  --output-dir out

Dry run (no UniVerse connection) to validate parallelism:

universe-reads --dry-run 300000 --count-only

Smoke-test the session factory mechanism (imports + kwargs filtering) without needing UniVerse:

PYTHONPATH=src:. python -m tests.test_dry_run --smoke-session-factory

Use as a library

from universe_reads import run


# Recommended: pass a module:function spec
results = run(
  file_name="READS",
  session_factory="my_uv:make_session",
  sessions=5,
  count_only=True,
)

print(results)

Dry-run (no UniVerse needed):

from universe_reads import run


results = run(file_name="", dry_run=300000)

Arguments

CLI (universe-reads)

• --file (default: UV_FILE): UniVerse file name to read
• --session-factory (required unless --dry-run): session factory in the form module.sub:make_session
• --sessions (default: 5): number of concurrent sessions/processes
• --count-only (default: off, but implied when no --output-dir): count records only
• --output-dir (default: unset): write one worker-N.ndjson per worker
• --batch-size (default: 500): number of keys per worker IPC batch (also the batch read size)
• --progress-every (default: 5.0): seconds between throughput logs
• --max-records (default: unset): debug cap (not a strict global cap)
• --dry-run N (default: unset): do not connect; generate N fake keys/records

Retry behaviour (for transient RPC / network failures):

• --retries (default: 3): number of retries per batch/key before giving up. 0 disables retries
• --retry-backoff (default: 0.5): base delay in seconds between retries (doubles each attempt: 0.5s, 1s, 2s)

Pooling hints (optional; passed as kwargs to the session factory if it supports them):

• --pooling / --no-pooling (default: neither): enable/disable pooling hints
• --min-pool-size (default: unset): minimum pool size hint
• --max-pool-size (default: unset): maximum pool size hint

Library (universe_reads.run)

run() accepts the same knobs as the CLI, as keyword args:

• file_name
• session_factory (spec string or top-level callable; optional when dry_run is set)
• sessions=5
• count_only=True
• output_dir=None
• batch_size=500
• progress_every_seconds=5.0
• max_records=None
• dry_run=None
• retries=3, retry_backoff=0.5
• pooling_on=None, min_pool_size=None, max_pool_size=None

Built-in env-based session factory

If you use:

--session-factory universe_reads.session_factory:make_session

It supports these environment variables:

• Required: UV_USER, UV_PASSWORD
• Optional: UV_HOST, UV_ACCOUNT, UV_PORT, UV_TIMEOUT, UV_SERVICE, UV_ENCODING, UV_SSL
• Optional pooling hints: UV_POOLING_ON, UV_MIN_POOL_SIZE, UV_MAX_POOL_SIZE

Where to plug in real uopy calls

All UniVerse file I/O calls are isolated in:

• src/universe_reads/uopy_adapter.py

You manage session creation via --session-factory. The adapter implements:

• iter_keys() (select list key streaming)
• read_record() / read_records()