python-peewee
Scannednpx machina-cli add skill narumiruna/agent-skills/python-peewee --openclawPython Peewee
Use Peewee with DatabaseProxy and scoped connection/transaction patterns.
Set up
DatabaseProxy & BaseModel
from peewee import DatabaseProxy, Model
db_proxy = DatabaseProxy()
class BaseModel(Model):
class Meta:
database = db_proxy
Initialize DB
from peewee import SqliteDatabase
db = SqliteDatabase("app.db", pragmas={"foreign_keys": 1})
db_proxy.initialize(db)
Use connections and transactions
Read (no transaction)
with db_proxy.obj.connection_context():
rows = MyModel.select().limit(100)
Write (atomic)
with db_proxy.obj.atomic():
a.save()
b.save()
Combined
db = db_proxy.obj
with db.connection_context():
with db.atomic():
...
Use connection_context() for scoped connections (open/close).
Use atomic() for atomic writes (BEGIN/COMMIT/ROLLBACK).
Test with SQLite
import pytest
from peewee import SqliteDatabase
@pytest.fixture
def test_db(tmp_path):
db = SqliteDatabase(str(tmp_path / "test.db"))
db_proxy.initialize(db)
with db.connection_context():
db.create_tables([MyModel])
yield db
db.close()
Source
git clone https://github.com/narumiruna/agent-skills/blob/main/skills/python-peewee/SKILL.mdView on GitHub Overview
This skill teaches using Peewee ORM with a DatabaseProxy and scoped connection/transaction patterns. It covers setting up a proxied database, managing connection_context and atomic blocks, and testing with SQLite to keep tests fast and isolated.
How This Skill Works
A DatabaseProxy acts as a placeholder database for Peewee models. You initialize it with a concrete database (e.g., SqliteDatabase) and attach it to a BaseModel via Meta.database. Reads run inside connection_context() for scoped connections, while writes run inside atomic() for transactional safety; you can nest these contexts as needed.
When to Use It
- Setting up a DatabaseProxy during app boot and wiring BaseModel to use the proxy
- Performing scoped reads using connection_context() to open/close connections
- Wrapping related writes in an atomic() block to ensure transactional integrity
- Combining nested contexts (e.g., with db.connection_context(): with db.atomic(): ...) for complex operations
- Writing tests with SQLite to keep tests fast, isolated, and easy to reset
Quick Start
- Step 1: Create a DatabaseProxy and a BaseModel that uses it
- Step 2: Initialize the proxy with a SqliteDatabase and call db_proxy.initialize(db)
- Step 3: Use with db_proxy.obj.connection_context() for reads or with db_proxy.obj.atomic() for writes
Best Practices
- Create a single DatabaseProxy per app and initialize it with a real database instance
- Use with db_proxy.obj.connection_context() for scoped reads/writes
- Wrap multiple writes in with db_proxy.obj.atomic() to ensure atomicity
- Nest contexts to reflect operation flow (e.g., connection_context inside atomic or vice versa)
- When testing, use a temporary SQLite database and clean up after tests
Example Use Cases
- App startup defines db_proxy and a BaseModel.Meta.database = db_proxy
- Read operation within a scoped context: with db_proxy.obj.connection_context(): rows = Model.select().limit(100)
- Batch writes inside a single atomic block: with db_proxy.obj.atomic(): a.save(); b.save()
- Combined pattern: with db_proxy.obj.connection_context(): with db_proxy.obj.atomic(): ...
- Pytest fixture for tests: initialize a SqliteDatabase in a temp path, wire via db_proxy, create tables, yield, then close
