oceanic-mysql-orm 1.0.4

A production-grade async MySQL ORM with LoopBack-style nested relation queries, typed model instances, and zero N+1.

oceanic-mysql-orm

A production-grade async MySQL connector with LoopBack-style nested relation queries and fully typed model instances. Built for FastAPI.

---

Features

• Declarative models — define once, auto-registered
• LoopBack-style filter DSL (where, order, limit, skip, fields, include)
• Nested relation loading with zero N+1 queries (batch IN-clause strategy)
• Typed model instances returned — full autocomplete, no raw dicts
• RelationNotLoadedError if you access an unloaded relation — no silent bugs
• All 4 relation types: hasOne, hasMany, belongsTo, hasManyThrough
• Scoped includes — filter/sort/limit inside each relation independently
• Async connection pool (aiomysql)
• Transaction support
• Raw SQL escape hatch

---

Install

pip install oceanic-mysql-orm

Or from source:

git clone https://gitlab.com/MaheshPulivarthi181/mysql-orm-connector.git
cd mysql-orm-connector
pip install .

For a detailed guide, see USAGE.md.

---

Quick Start

1. Define your models

# models.py
from mysql_connector import MySQLModel, Field, Relation

class User(MySQLModel):
    __table__ = "users"

    id:     int           = Field(primary_key=True)
    name:   str           = Field(max_length=100)
    email:  str           = Field(unique=True)
    active: bool          = Field(default=True)

    orders:  list["Order"]  = Relation(type="hasMany", foreign_key="user_id")
    profile: "Profile"      = Relation(type="hasOne",  foreign_key="user_id")

class Order(MySQLModel):
    __table__ = "orders"

    id:      int    = Field(primary_key=True)
    user_id: int    = Field(foreign_key="users.id")
    total:   float
    status:  str    = Field(default="pending")

    user:  "User"           = Relation(type="belongsTo", foreign_key="user_id")
    items: list["OrderItem"] = Relation(type="hasMany",  foreign_key="order_id")

class OrderItem(MySQLModel):
    __table__ = "order_items"

    id:         int   = Field(primary_key=True)
    order_id:   int   = Field(foreign_key="orders.id")
    product_id: int   = Field(foreign_key="products.id")
    qty:        int   = Field(default=1)
    unit_price: float

    product: "Product" = Relation(type="belongsTo", foreign_key="product_id")

2. Setup the connector

from mysql_connector import MySQLConnector
import models  # importing triggers auto-registration

connector = MySQLConnector(
    host="localhost", database="mydb",
    user="root", password="secret",
)
await connector.connect()

3. Query with nested relations

# Simple include
users = await connector.find(User, {
    "include": ["orders"],
})

# Deep nested — 3 levels, only 4 SQL queries total
users = await connector.find(User, {
    "where":   {"active": True},
    "include": ["orders.items.product"],
    "limit":   10,
})
for user in users:
    print(user.name)
    for order in user.orders:
        for item in order.items:
            print(f"  {item.product.name} x{item.qty}")

# Scoped include — filter inside the relation
users = await connector.find(User, {
    "include": [
        {
            "relation": "orders",
            "scope": {
                "where":   {"status": "paid"},
                "order":   ["created_at DESC"],
                "limit":   5,
                "include": [
                    {
                        "relation": "items",
                        "scope": {
                            "where":   {"qty": {"gt": 0}},
                            "include": ["product"],
                        }
                    }
                ]
            }
        }
    ]
})

---

Filter DSL Reference

filter = {
    # WHERE conditions
    "where": {
        "status":  "active",                    # equality
        "age":     {"gt": 18},                  # >
        "price":   {"gte": 100, "lte": 500},    # BETWEEN-style
        "name":    {"like": "%raj%"},            # LIKE
        "role":    {"inq": ["admin", "user"]},  # IN
        "deleted": {"neq": True},               # !=
        "or": [                                 # OR
            {"status": "paid"},
            {"status": "refunded"},
        ],
    },
    "order":   ["name ASC", "created_at DESC"], # ORDER BY
    "limit":   20,                              # LIMIT
    "skip":    40,                              # OFFSET
    "fields":  ["id", "name", "email"],         # SELECT specific columns
    "include": ["orders.items.product"],        # nested relations
}

Supported operators

Operator	SQL equivalent
eq	= %s
neq	!= %s
gt	> %s
gte	>= %s
lt	< %s
lte	<= %s
like	LIKE %s
nlike	NOT LIKE %s
inq	IN (...)
nin	NOT IN (...)
between	BETWEEN %s AND %s
regexp	REGEXP %s

---

CRUD Reference

# Find many
users = await connector.find(User, filter)

# Find one
user = await connector.find_one(User, {"where": {"email": "..."}})

# Find by PK
user = await connector.find_by_id(User, 42, {"include": ["orders"]})

# Create
user = await connector.create(User, {"name": "Raj", "email": "[e-mail]"})

# Update by PK — returns updated instance
user = await connector.update_by_id(User, 42, {"name": "Rajesh"})

# Update many
count = await connector.update_all(User, where={"active": False}, data={"deleted": True})

# Delete by PK
deleted = await connector.delete_by_id(User, 42)  # bool

# Delete many
count = await connector.delete_all(User, {"active": False})

# Count
n = await connector.count(User, {"active": True})

# Exists
exists = await connector.exists(User, {"email": "[e-mail]"})

# Raw SQL
rows = await connector.raw("SELECT * FROM users WHERE id = %s", [1])

# Transaction
async with connector.transaction() as conn:
    async with conn.cursor(aiomysql.DictCursor) as cur:
        await cur.execute("INSERT INTO users ...")
        await cur.execute("INSERT INTO orders ...")

---

FastAPI integration

from fastapi import FastAPI
from mysql_connector import MySQLConnector
import models   # auto-registers all models

connector = MySQLConnector(host="localhost", database="mydb", ...)

@asynccontextmanager
async def lifespan(app):
    await connector.connect()
    yield
    await connector.disconnect()

app = FastAPI(lifespan=lifespan)

@app.get("/users/{id}")
async def get_user(id: int):
    user = await connector.find_by_id(User, id, {
        "include": ["orders.items.product", "profile"]
    })
    if not user:
        raise HTTPException(404)
    return user.to_dict()   # fully serialized with nested objects

---

Project Structure

src/mysql_connector/
├── __init__.py           # public API exports
├── connector.py          # MySQLConnector — main entry point
├── model.py              # MySQLModel base class + ModelMeta metaclass
├── fields.py             # Field() and Relation() descriptors
├── registry.py           # global model registry (singleton)
├── pool.py               # async connection pool (aiomysql)
├── query_builder.py      # filter DSL → parameterized SQL
├── relation_resolver.py  # batch relation loading (no N+1)
├── types.py              # FilterOptions, FieldDefinition, etc.
└── exceptions.py         # typed exceptions