Module Proposal (secondary option): moschitta-orm

[skyler-saville]

Overview

This proposal outlines the implementation of the moschitta-orm module for the Moschitta Framework. The goal is to create a simple and intuitive ORM that functions similarly to SQLAlchemy but is more lightweight and only uses the Python standard library. The ORM will be asynchronous and will primarily support SQLite, with potential extension to other databases such as Postgres.

Objectives

1. Create an asynchronous ORM using the Python standard library.
2. Use Poetry to manage dependencies and the module's folder structure.
3. Implement a Model class that users can extend to create their own models.
4. Follow best practices and Pythonic principles, including the use of docstrings, type hints, and decorators.

Folder Structure

The folder structure for the moschitta-orm module will be as follows:

moschitta-orm/
├── moschitta_orm/
│   ├── __init__.py
│   ├── base.py
│   ├── database.py
│   ├── models.py
│   ├── utils.py
│   └── decorators.py
├── tests/
│   ├── __init__.py
│   └── test_orm.py
├── pyproject.toml
└── README.md

Implementation Details

1. BaseModel Class

moschitta_orm/base.py

from typing import Any, Dict, Type, TypeVar

T = TypeVar('T', bound='BaseModel')

class BaseModel:
    """Base class for all models."""
    def __init__(self, **kwargs):
        for key, value in kwargs.items():
            setattr(self, key, value)
    
    def to_dict(self) -> Dict[str, Any]:
        """Convert the model instance to a dictionary."""
        return {key: getattr(self, key) for key in self.__annotations__.keys()}

    @classmethod
    def from_dict(cls: Type[T], data: Dict[str, Any]) -> T:
        """Create a model instance from a dictionary."""
        return cls(**data)

2. Database Class

moschitta_orm/database.py

import sqlite3
import asyncio
from typing import List, Dict, Any

class Database:
    """Database connection and query handling."""
    def __init__(self, db_url: str):
        self.db_url = db_url

    async def execute(self, query: str, params: List[Any] = []):
        """Execute a query asynchronously."""
        await asyncio.sleep(0)  # Simulate async operation
        with sqlite3.connect(self.db_url) as conn:
            cursor = conn.cursor()
            cursor.execute(query, params)
            conn.commit()
    
    async def fetchall(self, query: str, params: List[Any] = []) -> List[Dict[str, Any]]:
        """Fetch all results from a query asynchronously."""
        await asyncio.sleep(0)  # Simulate async operation
        with sqlite3.connect(self.db_url) as conn:
            cursor = conn.cursor()
            cursor.execute(query, params)
            rows = cursor.fetchall()
            columns = [desc[0] for desc in cursor.description]
            return [dict(zip(columns, row)) for row in rows]

3. Sample Model

moschitta_orm/models.py

from moschitta_orm.base import BaseModel
from moschitta_orm.decorators import register_model

@register_model
class User(BaseModel):
    id: int
    name: str
    email: str

4. Utility Functions

moschitta_orm/utils.py

from typing import Dict
from moschitta_orm.database import Database

async def create_table(db: Database, table_name: str, columns: Dict[str, str]):
    """Create a table with the given name and column definitions."""
    column_defs = ", ".join(f"{name} {dtype}" for name, dtype in columns.items())
    query = f"CREATE TABLE IF NOT EXISTS {table_name} ({column_defs});"
    await db.execute(query)

5. Model Registration Decorator

moschitta_orm/decorators.py

from typing import Type

models_registry = {}

def register_model(cls: Type):
    """Decorator to register a model class."""
    models_registry[cls.__name__] = cls
    return cls

Example Usage

# example.py
import asyncio
from moschitta_orm.database import Database
from moschitta_orm.models import User
from moschitta_orm.utils import create_table

async def main():
    db = Database('example.db')
    
    # Create the users table
    await create_table(db, 'users', {
        'id': 'INTEGER PRIMARY KEY',
        'name': 'TEXT',
        'email': 'TEXT'
    })

    # Create a new user instance
    user = User(id=1, name='John Doe', email='[email protected]')
    
    # Insert user into the database
    await db.execute("INSERT INTO users (id, name, email) VALUES (?, ?, ?)", [user.id, user.name, user.email])
    
    # Fetch all users from the database
    users = await db.fetchall("SELECT * FROM users")
    for user_data in users:
        user = User.from_dict(user_data)
        print(user.to_dict())

# Run the main function
asyncio.run(main())

Testing

Add tests in the tests folder to ensure the ORM works as expected.

tests/test_orm.py

import asyncio
import pytest
from moschitta_orm.database import Database
from moschitta_orm.models import User
from moschitta_orm.utils import create_table

@pytest.mark.asyncio
async def test_create_and_fetch_user():
    db = Database(':memory:')  # Use in-memory database for testing
    
    # Create the users table
    await create_table(db, 'users', {
        'id': 'INTEGER PRIMARY KEY',
        'name': 'TEXT',
        'email': 'TEXT'
    })

    # Create a new user instance
    user = User(id=1, name='John Doe', email='[email protected]')
    
    # Insert user into the database
    await db.execute("INSERT INTO users (id, name, email) VALUES (?, ?, ?)", [user.id, user.name, user.email])
    
    # Fetch all users from the database
    users = await db.fetchall("SELECT * FROM users")
    assert len(users) == 1
    fetched_user = User.from_dict(users[0])
    assert fetched_user.id == 1
    assert fetched_user.name == 'John Doe'
    assert fetched_user.email == '[email protected]'

Conclusion

This proposal outlines the basic structure and implementation of the moschitta-orm module. By following this plan, we can create a simple, efficient, and Pythonic ORM for the Moschitta Framework that is easy to use and extend.