Kosmos-Py

Kosmos-Py is a high-performance, asynchronous MongoDB ORM/ODM-like framework and data persistence layer for Python. It is designed to establish a consistent, type-safe environment for backend services, combining Pydantic models with MongoDB operations, pandas/polars/pyarrow data science integration, secret resolution, timezone-aware timeframing.

---

Table of Contents

1. Core Philosophy & Architecture
2. Ignition & Configuration
3. Liminal Configuration & Secrets
4. Defining Models & Lifecycles
5. Atomic Counters
6. Field Transformers & Custom Metadata
7. Querying and Filtering (MongoDetector)
8. Data Saving & Recording (MongoRecorder)
9. GridFS Blobs (PersistableBlob)
10. Time & Profiling Utilities
11. Release Notes

---

1. Core Philosophy & Architecture

• State of Matter (ModelState):
  • Unset: The model exists only in memory and has not been prepared for database synchronization.
  • Transition: The model is collapsing its changes into a database instruction representation.
  • Material: The model is fully synchronized/entangled with the database and is considered stable.
• Collapse (collapse()): Compiles the current attributes of a model instance (including auto-timestamps, atomic increments, etc.) into a database update instruction representation (Ripple).
• Decohere (decohere()): Finalizes the model's state in-memory after database update confirmation (e.g. converting a delta increment to a static value, setting IDs).

---

2. Ignition & Configuration

Before interacting with databases or configuration properties, you must ignite the environment using ignite.

import kosmos as km

# Load environment variables and verify the read connection
km.ignite("app.env")

The configuration is mapped to connection affinities using PurposeAffinity:

• PurposeAffinity.Detector: Collection query / read access (uses MONGODB_DETECTOR_URI).
• PurposeAffinity.Recorder: Collection write / update access (uses MONGODB_RECORDER_URI).
• PurposeAffinity.Creator: Collection metadata / index creation (uses MONGODB_CREATOR_URI).
• PurposeAffinity.Admin: Database / administrative access (uses MONGODB_ADMIN_URI).

The affinity is chosen by the access path, not by the caller: reads (detect(...), open_blob) connect as Detector, and writes (record, Model.update(...), merge_into) connect as Recorder.

Every variable except MONGODB_URI is optional. An unset purpose falls back through progressively broader ones, ending at MONGODB_URI, so configuring a single URI is a supported deployment:

Purpose	Resolution order
Detector	MONGODB_DETECTOR_URI → MONGODB_RECORDER_URI → MONGODB_URI
Recorder	MONGODB_RECORDER_URI → MONGODB_URI
Creator	MONGODB_CREATOR_URI → MONGODB_ADMIN_URI → MONGODB_URI
Admin	MONGODB_ADMIN_URI → MONGODB_URI

A purpose only borrows a credential that can already do its job — a recorder can read, an admin can create indexes — so a borrowed connection is more privileged than the purpose needs, never less. Purposes that resolve to the same URI share one underlying client, so a single-URI deployment opens no extra connections.

---

3. Liminal Configuration & Secrets

Kosmos-Py provides a configuration mapper called LiminalStructure that matches environment variables to Python dataclass fields using the MapStruct annotation.

Liminal Configuration Dataclass

from dataclasses import dataclass
from typing import Annotated
import kosmos as km

@dataclass
class CustomAppConstants:
    app_port: Annotated[str, km.ether.MapStruct("PORT")] = "8080"
    db_name: Annotated[str, km.ether.MapStruct("DB_NAME")] = "default_db"

# Wrap inside LiminalStructure to enable lazy environment resolution
liminal_constants = km.ether.struct.LiminalStructure(CustomAppConstants())

# Retrieve resolved constants
constants = liminal_constants.collapse()
print(constants.db_name)

Secrets Resolution

If strings in your environment configuration follow the __secret:SECRET_NAME:VERSION__ pattern, Kosmos-Py will automatically intercept them and query Google Cloud Secret Manager via GCPSecretManager.

# Example Env File: app.env
PROJECT_ID=my-gcp-project-123
MONGODB_URI=mongodb://user:__secret:mongodb-prod-password:latest__@localhost:27017/prod

When ignite parses the URI, the secret is resolved invisibly to the caller, and credentials are masked during standard logging.

---

4. Defining Models & Lifecycles

The base hierarchy is a ladder, and you pick the rung that matches what you are modelling:

Base	Adds	Use for
Form	document round-tripping (from_doc/dump_doc), the q/qs/oq query namespaces, dirty tracking — no _id	read-only shapes: projections and aggregation results whose _id is a $group key rather than an ObjectId
Model	id: PyObjectId | None mapped to _id, plus collapse_id()/has_id()	anything with a document identity
ParticleBase	the persistence state machine (Unset → Transition → Material) and DB metadata	models you persist
Persistable	automatic created_at / updated_time timestamps	the common case — start here

Most application models subclass Persistable. Reach for Form only when a shape is never written back and its _id is not an ObjectId:

from pydantic import Field
import kosmos as km

@km.declare_persist_db(db_name="store_db", collection_name="sales", version=1)
class Sale(km.Persistable):
    region: str
    amount: float

# A $group key is arbitrary BSON, not an ObjectId — so the target is a Form.
class GroupedSales(km.Form):
    id: str = Field(alias="_id")     # the $group key, e.g. "west"
    total: float

# Field names in aggregation helpers are bare — kosmos adds the "$" prefix.
pipeline = km.detect(Sale).group_by(km.pth("region")).acc(km.fld("total").with_sum("amount"))
rows = km.Projector(GroupedSales).of(pipeline).load_many()

To declare database mappings, use the @declare_persist_db decorator.

from bson import ObjectId
from pydantic import Field
import kosmos as km

@km.declare_persist_db(db_name="store_db", collection_name="products", version=1)
class Product(km.Persistable):
    name: str = Field(alias="title")
    price: float
    category_id: ObjectId | None = None

Lifecycle & Update Tracking

When a model is instantiated directly, it is marked as modified (has_update = True). When loaded from the database, it is clean (has_update = False), and assigning any declared field marks it modified again automatically. To force-mark a model without mutating it, use mark_updated().

# Query database
detector = km.MongoDetector[Product](Product)
product = detector.filter(km.fld("title") == "Laptop").load_one()

# Modifying a field automatically marks the model as updated
product.price = 1200.00
assert product.has_update

# Save modifications back to MongoDB
km.record(product)

---

5. Atomic Counters

For concurrent systems, Kosmos-Py provides IncrCounter (aliased to IntCounter / ZeroCounter), which compiles to MongoDB $inc operations.

import kosmos as km
from pydantic import Field

@km.declare_persist_db(db_name="store_db", collection_name="stock_inventory")
class Inventory(km.Persistable):
    item_id: str
    stock: km.IncrCounter = Field(default=km.ZeroCounter)

# Initialize Inventory
inv = Inventory(item_id="sku-102")
km.record(inv)

# Load and increment atomically
detector = km.detect(Inventory)
loaded_inv = detector.filter(km.fld("item_id") == "sku-102").load_one()

loaded_inv.stock += 5  # Queues an atomic increment of +5
km.record(loaded_inv)  # Sends {"$inc": {"stock": 5}} to MongoDB

---

6. Field Transformers & Custom Metadata

Kosmos-Py uses type annotations to hook lifecycle transformations into Pydantic models.

Annotation / Type	Target Operations	Action
StrUpper	Initialization / Set	Converts string to uppercase
StrLower	Initialization / Set	Converts string to lowercase
TimeInserted	Document Creation	Automatically sets current UTC timestamp once on creation
TimeUpdated	Document Update	Sets current UTC timestamp on every save

Example Hook Usage:

import kosmos as km
from typing import Annotated

@km.declare_persist_db(db_name="analytics_db", collection_name="logs")
class LogEntry(km.ParticleBase):
    log_level: km.meta.annotation.StrUpper  # Always converts e.g. "info" to "INFO"
    message: str
    created_at: km.meta.annotation.TimeInserted  # Auto set on insert
    updated_at: km.meta.annotation.TimeUpdated   # Auto updated on save

---

7. Querying and Filtering (MongoDetector)

The MongoDetector class (instantiated with km.detect(Model)) exposes a chainable query builder supporting projection, filtering, pagination, lookups, aggregation, and conversion to standard scientific formats.

Fluent Aggregation & Querying

import kosmos as km

detector = km.detect(Product)

# Chain filters and lookup pipelines
results = (
    detector
    .filter((km.fld("price") > 100) & (km.fld("category") == "electronics"))
    .sort("price", descending=True)
    .skip(10)
    .limit(5)
    .load_many()
)

Type Resolution & Aliasing

Using fld() automatically resolves your Pydantic alias fields to their DB representation.

# In Product, "name" is aliased to "title"
# km.fld("name") compiles to query field: "title"
product = detector.filter(km.fld("name") == "UltraBook").load_one()

Loading to Pandas, Polars, and PyArrow

Kosmos-Py natively outputs your query results as data science containers using pymongoarrow:

# Load query results directly to Pandas
df_pandas = detector.filter(km.fld("price") > 50).load_dataframe()

# Load query results directly to Polars
df_polars = detector.filter(km.fld("price") > 50).load_polars()

# Load query results directly to PyArrow Table
arrow_table = detector.filter(km.fld("price") > 50).load_table()

Async Operations

All loading and aggregation pipeline methods support async execution:

# Async load query
await detector.filter(km.fld("price") > 500).load_one_async()
await detector.filter(km.fld("price") > 500).load_many_async()

---

8. Data Saving & Recording (MongoRecorder)

Saving and updating operations are handled by MongoRecorder or the wrapper functions km.record(obj) and km.record_async(obj).

Single Document Saving

# Sync saving
km.record(product)

# Async saving
await km.record_async(product)

Dataframe Bulk Operations

You can bulk insert or upsert DataFrames containing hundreds of rows using bulk write queries.

import pandas as pd
import kosmos as km

recorder = km.MongoRecorder(Product)
df = pd.DataFrame([
    {"title": "Tablet A", "price": 299.99},
    {"title": "Tablet B", "price": 499.99}
])

# Insert bulk rows into database
recorder.insert_dataframe(df)

# Upsert dataframes based on specific keys
recorder.update_dataframe(df, on=["title"], upsert=True)

---

9. GridFS Blobs (PersistableBlob)

For storing large files, inherit from PersistableBlob and specify is_blob=True in your DB declaration. Kosmos-Py manages upload, metadata association, and deletes old chunks from GridFS automatically when files are replaced.

from typing import Optional
import io
import kosmos as km

@km.declare_persist_db(db_name="files_db", collection_name="attachments", is_blob=True)
class DocumentAttachment(km.PersistableBlob):
    data: bytes = b""
    metadata: Optional[dict] = None

    def dump_buffer(self) -> io.BytesIO:
        return io.BytesIO(self.data)

# 1. Upload a new file
new_attachment = DocumentAttachment(
    filename="invoice.pdf",
    data=b"Raw PDF bytes content",
    metadata={"customer_id": "1002"}
)
km.record(new_attachment)  # Uploads to GridFS

# 2. Retrieve file content
detector = km.detect(DocumentAttachment)
loaded = detector.filter(km.fld("filename") == "invoice.pdf").load_one()

# Open download stream to read contents
stream = km.open_blob(loaded)
file_bytes = stream.read()
print(file_bytes)

Both sync (open_blob, record) and async (open_blob_async, record_async) methods are supported.

---

10. Time & Profiling Utilities

Timeframes and Chronological Alignment

Kosmos-Py has a robust TimeFrame utility mapping to specific time bounds (hourly, daily, weekly, monthly, quarterly, yearly).

from datetime import datetime, timezone
import kosmos.time as kt

moment = datetime(2026, 6, 1, 15, 0, 0, tzinfo=timezone.utc)

# Align date to Daily bounds
day_frame = kt.DailyFrame.create(moment=moment, tzone=timezone.utc)
print(day_frame.floor)    # 2026-06-01 00:00:00+00:00
print(day_frame.ceiling)  # 2026-06-02 00:00:00+00:00

# Jump to previous time frames
yesterday = day_frame.get_previous_frame()
last_week = day_frame.get_previous_x_frame(7)

Performance & Nested Profiling

Profile code blocks hierarchically using PerfTimer or the @timed decorator.

import kosmos.time as kt
import time

@kt.timed(name="Fetch API Data")
def fetch_api(ptimer=None):
    time.sleep(0.1)

@kt.timed(name="Process DB Records")
def process_db(ptimer=None):
    # Pass ptimer to record nested timer calls
    time.sleep(0.05)
    fetch_api(ptimer=ptimer)

# Run operations inside a parent timer scope
with kt.PerfTimer("Main Flow", verbose=True) as root_timer:
    process_db(ptimer=root_timer)
    
# Prints output similar to:
# Main Flow -> 1 times in 150.00 ms
# └── Process DB Records -> 1 times in 150.00 ms (100.0%)
# |  └── Fetch API Data -> 1 times in 100.00 ms (66.7%)

---

11. Release Notes

2.0.0 (breaking)

Full upgrade guide with rewrite tables: MIGRATION.md.

• km.fld / Model.q are strict about a bare None. Comparing with a bare None now raises ValueError instead of skipping the filter, so an unset config value or a failed lookup can no longer silently widen a query into a full-collection match. The lenient behavior moved to km.ofld / Model.oq (LenientQueryableField). km.sfld / Model.qs remain permanent synonyms of the strict default, so code that adopted strict early needs no change. To compare against null, stay explicit: fld("x") == km.lit(None) or is_null(). is_within(None) and is_exists(None) still skip on every field class — their Optional signatures make that the contract.
• Model split into Form + identity. Form carries document round-tripping and the query namespaces with no _id; Model(Form) adds id: PyObjectId | None. Projector, MongoDetector, GroupDetector, and detect() are now bounded on Form, so a projection target can declare the _id it actually has — an aggregation $group key is arbitrary BSON, not an ObjectId. This is a widening: every existing Model still satisfies it, so no consumer change is required. See §4.
• Projector.Project() / .From() renamed to .project() / .of(). The old spellings were removed rather than aliased — keeping .From would have preserved the Form/From transposition the rename existed to remove. This is the one mandatory rewrite in 2.0; it is mechanical (grep .Project( and .From().

1.7.0

Changes driven by downstream consumer feedback (see notes/kosmos_improvement_requests.md).

Superseded by 2.0.0: the None-semantics and StrictQueryableField bullets below describe 1.7.0 behavior. In 2.0 the default inverted — fld raises and ofld skips. Read the 2.0.0 notes above for current behavior.

Behavior changes

• Debug pipeline printing removed. Aggregation pipelines are no longer printed to stdout on every query; they are logged at DEBUG level on the kosmos.mongo.detector logger instead.
• Mutations mark models updated. Assigning a declared field on an initialized model now sets has_update/should_persist automatically; mark_updated() remains for force-marking.
• None semantics are now uniform across the query DSL. A bare None on any comparison (==, !=, >, <, >=, <=, is_in, is_not_in, is_all, is_not_all) skips the filter (empty, match-all predicate), so optional filters pass straight through. To genuinely compare against null, be explicit: fld("x") == km.lit(None) / != km.lit(None), or the named is_null() / is_not_null(). Breaking: == None previously meant is_null(). Also fixed: lit(...) values were double-wrapped on every operator except == (leaking raw LiteralInput objects into queries); lit() literals are now field-linked so query-input transformers (e.g. StrUpper) apply to them, except lit(None) which always passes through as null.
• StrictQueryableField / km.sfld for call sites where a None would be a bug rather than an optional filter: identical to fld except any bare None comparison raises ValueError instead of skipping. Predicates from strict and lenient fields combine freely with &/|. Strict handling becomes the default in a future major release — see MIGRATION.md for the phased plan.
• Explicitly cleared fields are $unset. Setting a field to None on a loaded model now removes it from the document on persist (using pydantic's model_fields_set to distinguish "never set" from "set to None"). dump_doc(include_none=True) is available to serialize Nones as null.
• Bulk update chunking. write_bulk_unordered defaults to chunks of 1000 (was 10) and update_dataframe(_async) accepts chunk_size; both now return aggregated write counts (inserted/matched/modified/upserted/write_errors).
• DataFrame writes honor field annotations. insert_dataframe/update_dataframe apply NormalizeValue (e.g. StrUpper), refresh RefreshOnSet fields (e.g. updated_time), and route CoalesceOnInsert fields (e.g. created_at) into $setOnInsert on upserts.
• Typed startup exceptions. ignite() raises KosmosIgnitionError / KosmosConnectionError (both subclass KosmosError) instead of bare Exception.
• Dependencies. pyrefly and rich are dev-only dependencies now.

New APIs

• Typed query namespace: MyModel.q.age > 30 — unknown field names raise AttributeError at query-build time.
• Atomic partial updates: MyModel.update(km.fld("status") == "pending").set(status="paused").inc(tries=1).unset("error").exec(many=False, upsert=False) — alias resolution and NormalizeValue/RefreshOnSet semantics included; exec(many=True) with an empty filter requires allow_all=True.
• BSON-safe types: km.BsonTimedelta (stored as float seconds) and km.BsonDecimal (stored as Decimal128), alongside the existing PyObjectId/StrUpper family; TimeInserted/TimeUpdated are exported from kosmos now.
• Blob retention: @km.declare_persist_db(..., is_blob=True, blob_replace=False) keeps previously uploaded GridFS versions instead of deleting them when a blob is re-recorded. Note: the default (blob_replace=True) deletes the previous file, as before.