Metadata-Version: 2.4
Name: mydborm
Version: 1.8.0
Summary: Lightweight ORM for MySQL and YugabyteDB
Project-URL: Homepage, https://pypi.org/project/mydborm/
Project-URL: Repository, https://github.com/codengers/mydborm
Author-email: Codengers <info@codengers.com>
License: MIT
License-File: LICENSE
Keywords: database,mysql,orm,yugabytedb
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Database
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.8
Requires-Dist: mysql-connector-python>=8.0
Requires-Dist: psycopg2-binary>=2.9
Provides-Extra: async
Requires-Dist: aiomysql>=0.3; extra == 'async'
Requires-Dist: aiopg>=1.4; extra == 'async'
Provides-Extra: cli
Requires-Dist: rich>=13.0; extra == 'cli'
Requires-Dist: typer>=0.9; extra == 'cli'
Provides-Extra: dev
Requires-Dist: aiomysql>=0.3; extra == 'dev'
Requires-Dist: aiopg>=1.4; extra == 'dev'
Requires-Dist: hatchling; extra == 'dev'
Requires-Dist: mysql-connector-python>=8.0; extra == 'dev'
Requires-Dist: psycopg2-binary>=2.9; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.23; extra == 'dev'
Requires-Dist: pytest-cov; extra == 'dev'
Requires-Dist: pytest>=7; extra == 'dev'
Requires-Dist: ruff; extra == 'dev'
Provides-Extra: security
Requires-Dist: bcrypt>=4.0; extra == 'security'
Requires-Dist: cryptography>=41.0; extra == 'security'
Provides-Extra: ui
Requires-Dist: streamlit>=1.30; extra == 'ui'
Description-Content-Type: text/markdown

# mydborm

> **Lightweight Python ORM for MySQL and YugabyteDB.**
> Ship database-backed apps and data pipelines in minutes — not hours.

[![PyPI version](https://badge.fury.io/py/mydborm.svg)](https://pypi.org/project/mydborm/)
[![Python](https://img.shields.io/pypi/pyversions/mydborm)](https://pypi.org/project/mydborm/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)
[![Tests](https://github.com/codengers/mydborm/actions/workflows/ci.yml/badge.svg)](https://github.com/codengers/mydborm/actions)
[![Coverage](https://img.shields.io/badge/coverage-93%25-brightgreen)](https://github.com/codengers/mydborm)
[![PyPI Downloads](https://img.shields.io/pypi/dm/mydborm)](https://pypi.org/project/mydborm/)

---

## Why mydborm?

SQLAlchemy is powerful — and complex. Peewee is simple — but not distributed-SQL-aware. **mydborm sits in the middle**: zero-boilerplate Python classes, parameterized queries, bulk operations, and first-class [YugabyteDB](https://www.yugabyte.com/) support — all in a 47 KB package with no heavy dependencies.

| | mydborm | SQLAlchemy | Peewee |
|---|---|---|---|
| Install size | **47 KB** | 3 MB | 800 KB |
| MySQL support | ✅ | ✅ | ✅ |
| YugabyteDB (Distributed SQL) | ✅ Native | ⚠️ Workaround | ❌ |
| PostgreSQL | ✅ | ✅ | ✅ |
| Async built-in | ✅ | needs ext | ❌ |
| Password hashing | ✅ bcrypt | ❌ | ❌ |
| AES field encryption | ✅ Fernet | ❌ | ❌ |
| Soft delete mixin | ✅ | ❌ | ❌ |
| CLI included | ✅ | ❌ | ❌ |
| Learning curve | **Low** | High | Medium |

---

## Key Features

- **Lightweight & fast** — 47 KB, two runtime dependencies, sub-millisecond query overhead.
- **Native Distributed SQL** — purpose-built dialect for YugabyteDB (YSQL). Run the exact same model code against a single-node MySQL dev database and a distributed YugabyteDB production cluster.
- **Intuitive syntax** — one Python class = one table. No metaclass magic to learn, no session factories, no engine strings.
- **29 field types** — covers everything from `TinyIntField` to `EncryptedField` (AES-128) and `PasswordField` (bcrypt).
- **Bulk operations** — chunked `bulk_create / bulk_update / bulk_delete` with retry, exponential backoff, and progress callbacks. Built for data pipelines.
- **Mixins** — `SoftDeleteMixin`, `AuditMixin`, `TimestampMixin` drop into any model in one line.
- **Lifecycle hooks** — `before_create`, `after_create`, `before_update`, `after_update`, `before_delete`, `after_delete` with zero registration ceremony.
- **Async support** — `AsyncBaseModel` via `aiomysql` and `aiopg` for FastAPI and async microservices.
- **Security-first fields** — `PasswordField` auto-hashes on write; `EncryptedField` auto-encrypts. Plain text never stored.
- **Schema migrations** — auto-generate versioned SQL diff files. Run `mydborm generate --model myapp.models.User` from the CLI.

---

## Installation

```bash
# Core — MySQL + YugabyteDB + PostgreSQL
pip install mydborm

# Add CLI tools
pip install mydborm[cli]

# Add async support (FastAPI / asyncio)
pip install mydborm[async]

# Add security fields (bcrypt + AES)
pip install mydborm[security]

# Everything
pip install mydborm[cli,async,security]
```

**Runtime requirements:** Python 3.9+ · `mysql-connector-python` · `psycopg2-binary`

---

## Quick Start

### Scenario A — MySQL

Standard relational database. Works with any MySQL 8+ instance, including RDS and Cloud SQL.

```python
# ── quickstart_mysql.py ────────────────────────────────────────────────────
from mydborm import (
    db,
    BaseModel,
    IntField,
    StrField,
    BoolField,
    FloatField,
    EmailValidator,
)

# 1. Configure once — thread-safe, connection-pooled
db.configure(
    dialect  = "mysql",
    host     = "127.0.0.1",
    port     = 3306,
    user     = "root",
    password = "yourpassword",
    database = "shop",
    charset  = "utf8mb4",
)

# 2. Declare your model — one class, one table
class Product(BaseModel):
    __tablename__ = "products"
    id       = IntField(primary_key=True)
    name     = StrField(max_length=120, nullable=False)
    sku      = StrField(max_length=20,  nullable=False, unique=True)
    price    = FloatField(nullable=False)
    in_stock = BoolField(default=True)

# 3. Create the table (safe to call repeatedly — IF NOT EXISTS)
Product.create_table()

# 4. Insert a record — returns the new primary key (int)
pid = Product.create(
    name     = "Wireless Keyboard",
    sku      = "KB-WL-001",
    price    = 49.99,
    in_stock = True,
)
print(f"Created product #{pid}")   # Created product #1

# 5. Read it back
product = Product.get(id=pid)
print(product["name"])   # Wireless Keyboard

# 6. Query with filters
budget_items = (
    Product.query()
           .where("price__lte", 50.0)
           .where("in_stock", True)
           .order_by("price")
           .all()
)
print(f"Found {len(budget_items)} items under $50")

# 7. Update
Product.update({"price": 44.99}, id=pid)

# 8. Delete
Product.delete(id=pid)
```

---

### Scenario B — YugabyteDB (Distributed SQL)

Switch `dialect` and `port`. **Everything else is identical.** The same model, same CRUD calls, same query builder — now running against a horizontally scalable, fault-tolerant distributed cluster.

```python
# ── quickstart_yugabyte.py ─────────────────────────────────────────────────
from mydborm import db, BaseModel, IntField, StrField, BoolField, FloatField

# The ONLY change from MySQL: dialect + port
db.configure(
    dialect  = "yugabyte",       # ← changed
    host     = "127.0.0.1",
    port     = 5433,             # ← changed (YugabyteDB YSQL default)
    user     = "yugabyte",
    password = "yugabyte",
    database = "yugabyte",
)

# Same model — no changes needed
class Product(BaseModel):
    __tablename__ = "products"
    id       = IntField(primary_key=True)
    name     = StrField(max_length=120, nullable=False)
    sku      = StrField(max_length=20,  nullable=False, unique=True)
    price    = FloatField(nullable=False)
    in_stock = BoolField(default=True)

Product.create_table()

# mydborm handles the dialect differences internally:
#   MySQL      → AUTO_INCREMENT, backtick identifiers, TINYINT(1) for bool
#   YugabyteDB → SERIAL, double-quote identifiers, native BOOLEAN, JSONB, RETURNING id

pid = Product.create(
    name     = "Wireless Keyboard",
    sku      = "KB-WL-001",
    price    = 49.99,
    in_stock = True,
)

product = Product.get(id=pid)
print(product["name"])   # Wireless Keyboard — same API, distributed backend
```

**YugabyteDB dialect differences handled automatically by mydborm:**

| Feature | MySQL | YugabyteDB |
|---|---|---|
| Primary key | `AUTO_INCREMENT` | `SERIAL` |
| Identifiers | `` `backticks` `` | `"double-quotes"` |
| Boolean | `TINYINT(1)` | `BOOLEAN` |
| JSON | `JSON` | `JSONB` (indexable) |
| Timestamps | `DATETIME` | `TIMESTAMPTZ` |
| Insert return | `lastrowid` | `RETURNING id` |

---

## CRUD Reference

All examples use the `Product` model from the Quick Start above.

### Create

```python
# Single insert — returns new primary key (int)
pid = Product.create(name="Mouse", sku="MS-001", price=29.99, in_stock=True)

# Bulk insert — one SQL statement, returns row count
records = [
    {"name": f"Item {i}", "sku": f"SKU-{i:04d}", "price": float(i), "in_stock": True}
    for i in range(1, 1001)
]
count = Product.bulk_create(records)
print(f"Inserted {count} products")   # Inserted 1000 products

# Chunked bulk — for very large datasets with retry + progress
from mydborm.bulk import chunked_bulk_create

result = chunked_bulk_create(
    Product,
    records,
    chunk_size  = 500,
    retries     = 3,
    on_progress = lambda done, total: print(f"{done}/{total}"),
)
print(result.summary())
```

### Read

```python
# By primary key — returns ModelInstance or None
product = Product.get(id=1)

# Equality filter shorthand
in_stock = Product.filter(in_stock=True)

# Full query builder
results = (
    Product.query()
           .where("price__gte", 10.0)        # price >= 10
           .where("price__lte", 100.0)       # price <= 100
           .where("name__like", "%board%")   # LIKE
           .where("in_stock", True)
           .order_by("price", desc=False)
           .limit(20)
           .offset(0)
           .all()
)

# Aggregates
total    = Product.count(in_stock=True)
cheapest = Product.query().min("price")
avg      = Product.query().where("in_stock", True).avg("price")

# Check existence
exists = Product.exists(sku="KB-WL-001")   # True / False
```

**WHERE operators:**

| Syntax | SQL equivalent | Example |
|---|---|---|
| `"field", value` | `= %s` | `.where("in_stock", True)` |
| `"field__gt"` | `> %s` | `.where("price__gt", 50.0)` |
| `"field__lt"` | `< %s` | `.where("price__lt", 100.0)` |
| `"field__gte"` | `>= %s` | `.where("price__gte", 10.0)` |
| `"field__lte"` | `<= %s` | `.where("price__lte", 99.99)` |
| `"field__ne"` | `!= %s` | `.where("status__ne", "deleted")` |
| `"field__like"` | `LIKE %s` | `.where("name__like", "%board%")` |
| `"field__in"` | `IN (...)` | `.where("id__in", [1, 2, 3])` |
| `"field__null"` | `IS NULL / IS NOT NULL` | `.where("deleted_at__null", True)` |

### Update

```python
# Update by any field — returns rows affected (int)
rows = Product.update({"price": 39.99}, id=1)

# Update multiple rows
rows = Product.update({"in_stock": False}, price=0.0)

# Bulk update
updates = [{"id": i, "price": float(i) * 0.9} for i in range(1, 1001)]
Product.bulk_update(updates, key="id")
```

### Delete

```python
# Delete by any field — returns rows deleted (int)
deleted = Product.delete(id=1)

# Bulk delete by IDs
ids = [p["id"] for p in Product.filter(in_stock=False)]
Product.bulk_delete(ids)

# Soft delete (keeps row, sets deleted_at timestamp)
from mydborm.mixins import SoftDeleteMixin

class Post(BaseModel, SoftDeleteMixin):
    __tablename__ = "posts"
    id    = IntField(primary_key=True)
    title = StrField(max_length=200, nullable=False)

Post.soft_delete(id=1)            # sets deleted_at — row hidden from .all()
Post.restore(id=1)                # clears deleted_at — row visible again
Post.purge(id=1)                  # permanent delete
```

---

## Advanced Usage — Data Science & Pipelines

mydborm is well-suited for data extraction scripts and ingestion pipelines where you need to query a table, transform rows, and feed them into downstream tools (pandas, Polars, Kafka producers, etc.).

### Query → pandas DataFrame

```python
import pandas as pd
from mydborm import db, BaseModel, IntField, StrField, FloatField, DateTimeField

db.configure(dialect="mysql", host="127.0.0.1", port=3306,
             user="root", password="root", database="analytics")

class SalesEvent(BaseModel):
    __tablename__ = "sales_events"
    id         = IntField(primary_key=True)
    product_id = IntField(nullable=False)
    region     = StrField(max_length=50, nullable=False)
    revenue    = FloatField(nullable=False)
    created_at = DateTimeField(nullable=True)

# Pull last 30 days of high-value events
rows = (
    SalesEvent.query()
              .where("revenue__gte", 1000.0)
              .where("region__in", ["NA", "EU", "APAC"])
              .order_by("created_at", desc=True)
              .limit(10_000)
              .all()
)

# Convert to list of dicts — one line
records = [row.to_dict() for row in rows]

# Load into pandas
df = pd.DataFrame(records)
print(df.head())
print(df.groupby("region")["revenue"].sum())
```

### Bulk ingestion pipeline

```python
from mydborm.bulk import chunked_bulk_create
import csv

def ingest_csv(filepath: str, model_class, chunk_size: int = 500):
    """Load a CSV file into a mydborm model table — with retry and progress."""
    with open(filepath, newline="", encoding="utf-8") as f:
        reader  = csv.DictReader(f)
        records = list(reader)

    result = chunked_bulk_create(
        model_class,
        records,
        chunk_size  = chunk_size,
        retries     = 3,
        retry_delay = 0.5,
        on_progress = lambda done, total:
            print(f"\rIngesting... {done:,}/{total:,}", end=""),
    )
    print(f"\nDone. {result.inserted:,} rows inserted, {result.failed} failed.")
    if result.has_errors:
        for err in result.errors:
            print(f"  Chunk {err['chunk']}: {err['error']}")
    return result

ingest_csv("sales_2024.csv", SalesEvent, chunk_size=500)
```

### FastAPI microservice — async mode

```python
from fastapi import FastAPI, HTTPException
from contextlib import asynccontextmanager
from mydborm.async_db import async_db, AsyncBaseModel
from mydborm import IntField, StrField, FloatField, BoolField

class Product(AsyncBaseModel):
    __tablename__ = "products"
    id       = IntField(primary_key=True)
    name     = StrField(max_length=120, nullable=False)
    price    = FloatField(nullable=False)
    in_stock = BoolField(default=True)

@asynccontextmanager
async def lifespan(app: FastAPI):
    await async_db.configure(
        dialect="mysql", host="127.0.0.1", port=3306,
        user="root", password="root", database="shop"
    )
    await Product.create_table()
    yield
    await async_db.close()

app = FastAPI(lifespan=lifespan)

@app.get("/products/{product_id}")
async def get_product(product_id: int):
    product = await Product.get(id=product_id)
    if not product:
        raise HTTPException(status_code=404, detail="Not found")
    return product.to_dict()

@app.get("/products")
async def list_products(in_stock: bool = True):
    products = await Product.filter(in_stock=in_stock)
    return [p.to_dict() for p in products]

@app.post("/products")
async def create_product(name: str, price: float):
    pid = await Product.create(name=name, price=price, in_stock=True)
    return {"id": pid}
```

### YugabyteDB — distributed ingestion at scale

```python
# Switch dialect to target a YugabyteDB cluster.
# The chunked_bulk_create call is identical — mydborm handles
# ON CONFLICT DO UPDATE and RETURNING id automatically.

db.configure(
    dialect  = "yugabyte",
    host     = "yb-node-1.prod.internal",  # any node in the cluster
    port     = 5433,
    user     = "app_user",
    password = "securepassword",
    database = "analytics",
)

result = chunked_bulk_create(SalesEvent, records, chunk_size=1000)
print(result.summary())
# Operation : insert
# Total     : 500000
# Inserted  : 500000
# Failed    : 0
# Chunks    : 500
# Duration  : 42.3s
```

---

## Security Fields

```python
import os
from mydborm import db, BaseModel, IntField, StrField
from mydborm import PasswordField, EncryptedField

# pip install mydborm[security]

KEY = os.environ["ENCRYPTION_KEY"]   # generate with EncryptedField.generate_key()

class User(BaseModel):
    __tablename__ = "users"
    id         = IntField(primary_key=True)
    username   = StrField(max_length=50,  nullable=False)
    password   = PasswordField(nullable=False)              # bcrypt, auto-hashed
    api_secret = EncryptedField(secret_key=KEY)             # AES-128, auto-encrypted

User.create_table()

uid  = User.create(username="alice", password="hunter2", api_secret="my-api-secret-value")
user = User.get(id=uid)

# Password: verify only — never decryptable
ok = PasswordField.verify("hunter2", user["password"])   # True

# Encrypted field: decrypt when needed
plain = EncryptedField.decrypt(user["api_secret"], secret_key=KEY)   # my-api-secret-value
```

---

## CLI

```bash
pip install mydborm[cli]

# Test connection
mydborm ping --dialect mysql --port 3306 --user root --password root --database shop

# List tables with row counts
mydborm tables --dialect yugabyte --port 5433 --user yugabyte --password yugabyte --database analytics

# Inspect schema
mydborm inspect --dialect mysql --port 3306 --user root --password root --database shop

# Auto-generate migration file from model diff
mydborm generate \
  --dialect mysql --port 3306 --user root --password root --database shop \
  --model myapp.models.Product \
  --output migrations/ \
  --apply
```

---

## All Field Types

| Field | MySQL | YugabyteDB / PostgreSQL | Notes |
|---|---|---|---|
| `IntField` | `INT` | `INTEGER` | |
| `StrField(n)` | `VARCHAR(n)` | `VARCHAR(n)` | |
| `TextField` | `TEXT` | `TEXT` | |
| `BoolField` | `TINYINT(1)` | `BOOLEAN` | |
| `FloatField` | `FLOAT` | `FLOAT` | |
| `DecimalField(p,s)` | `DECIMAL(p,s)` | `DECIMAL(p,s)` | |
| `DateField` | `DATE` | `DATE` | |
| `DateTimeField` | `DATETIME` | `TIMESTAMP` | |
| `JSONField` | `JSON` | `JSONB` | indexable on YugabyteDB |
| `ForeignKeyField` | `INT` | `INTEGER` | |
| `TinyIntField` | `TINYINT` | `SMALLINT` | |
| `SmallIntField` | `SMALLINT` | `SMALLINT` | |
| `BigIntField` | `BIGINT` | `BIGINT` | |
| `UnsignedBigIntField` | `BIGINT UNSIGNED` | `NUMERIC(20)` | |
| `DoubleField` | `DOUBLE` | `DOUBLE PRECISION` | |
| `BitField(n)` | `BIT(n)` | `BIT(n)` | |
| `CharField(n)` | `CHAR(n)` | `CHAR(n)` | fixed-length |
| `TinyTextField` | `TINYTEXT` | `TEXT` | |
| `MediumTextField` | `MEDIUMTEXT` | `TEXT` | |
| `LongTextField` | `LONGTEXT` | `TEXT` | |
| `BinaryField(n)` | `BINARY(n)` | `BYTEA` | |
| `VarBinaryField(n)` | `VARBINARY(n)` | `BYTEA` | |
| `BlobField` | `BLOB` | `BYTEA` | |
| `TimeField` | `TIME` | `TIME` | |
| `TimestampField` | `TIMESTAMP` | `TIMESTAMPTZ` | |
| `EnumField(choices)` | `ENUM(...)` | `VARCHAR(n)` | |
| `SetField(choices)` | `SET(...)` | `TEXT[]` | |
| `PasswordField` | `VARCHAR(255)` | `VARCHAR(255)` | bcrypt, auto-hashed |
| `EncryptedField` | `TEXT` | `TEXT` | AES-128-CBC, auto-encrypted |

---

## Links

| Resource | URL |
|---|---|
| PyPI | https://pypi.org/project/mydborm/ |
| Documentation | https://codengers.github.io/mydborm/ |
| GitHub | https://github.com/codengers/mydborm |
| Issue tracker | https://github.com/codengers/mydborm/issues |
| Changelog | https://github.com/codengers/mydborm/blob/main/CHANGELOG.md |

---

## License & Contributing

**License:** MIT — free for commercial and personal use.

**Contributing:**
1. Fork the repository
2. Create a branch: `git checkout -b feature/your-feature`
3. Add tests — every PR must include tests
4. Run: `pytest tests/ -q` — must be all green
5. Open a PR to the `develop` branch

Bug reports and feature requests are welcome via [GitHub Issues](https://github.com/codengers/mydborm/issues).

---

*Built by [Atikrant Upadhye](https://github.com/codengers) · MIT License*
