| Name | commondao JSON |
| Version |
1.0.1
JSON |
| download |
| home_page | None |
| Summary | Mysql toolkit |
| upload_time | 2025-07-23 04:10:40 |
| maintainer | None |
| docs_url | None |
| author | None |
| requires_python | >=3.10 |
| license | Apache-2.0 |
| keywords |
mysql
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# CommonDAO
A powerful, type-safe, and Pydantic-integrated async MySQL toolkit for Python.
CommonDAO is a lightweight, type-safe async MySQL toolkit designed to simplify database operations with a clean, intuitive API. It integrates seamlessly with Pydantic for robust data validation while providing a comprehensive set of tools for common database tasks.
## ✨ Features
- **Async/Await Support**: Built on aiomysql for non-blocking database operations
- **Type Safety**: Strong typing with Python's type hints and runtime type checking
- **Pydantic Integration**: Seamless validation and transformation of database records to Pydantic models
- **SQL Injection Protection**: Parameterized queries for secure database access
- **Comprehensive CRUD Operations**: Simple methods for common database tasks
- **Raw SQL Support**: Full control when you need it with parameterized raw SQL
- **Connection Pooling**: Efficient database connection management
- **Minimal Boilerplate**: Write less code while maintaining readability and control
## 🚀 Installation
```bash
pip install commondao
```
## 🔍 Quick Start
```python
import asyncio
from commondao import connect
from pydantic import BaseModel
# Define your Pydantic model
class User(BaseModel):
id: int
name: str
email: str
async def main():
# Connect to database
config = {
'host': 'localhost',
'port': 3306,
'user': 'root',
'password': 'password',
'db': 'testdb',
'autocommit': True,
}
async with connect(**config) as db:
# Insert a new user
await db.insert('users', data={'name': 'John Doe', 'email': 'john@example.com'})
# Query the user with Pydantic model validation
user = await db.select_one(
"select * from users where email = :email",
User,
{"email": "john@example.com"}
)
print(f"User: {user.name} ({user.email})") # Output => User: John Doe (john@example.com)
if __name__ == "__main__":
asyncio.run(main())
```
## 📊 Core Operations
### Connection
```python
from commondao import connect
async with connect(
host='localhost',
port=3306,
user='root',
password='password',
db='testdb'
) as db:
# Your database operations here
pass
```
### Insert Data
```python
# Simple insert
await db.insert('users', data={'name': 'John', 'email': 'john@example.com'})
# Insert with ignore option (skips duplicate key errors)
await db.insert('users', data={'name': 'Jane', 'email': 'jane@example.com'}, ignore=True)
```
### Update Data
```python
# Update where id = 1
await db.update_by_key(
'users',
key={'id': 1},
data={'name': 'John Smith', 'email': 'john.smith@example.com'}
)
# Only non-None values will be updated
await db.update_by_key(
'users',
key={'id': 1},
data={'name': 'Jane Doe', 'email': None} # email won't be updated
)
```
### Delete Data
```python
# Delete where id = 1
await db.delete_by_key('users', key={'id': 1})
```
### Query Data
```python
# Get a single row
user = await db.get_by_key('users', key={'id': 1})
# Get a row or raise NotFoundError if not found
user = await db.get_by_key_or_fail('users', key={'id': 1})
# Use with Pydantic models
from pydantic import BaseModel
from commondao import RawSql
from typing import Annotated
class UserModel(BaseModel):
id: int
name: str
email: str
full_name: Annotated[str, RawSql("CONCAT(first_name, ' ', last_name)")]
# Query with model validation
user = await db.select_one(
"select * from users where id = :id",
UserModel,
{"id": 1}
)
# Query multiple rows
users = await db.select_all(
"select * from users where status = :status",
UserModel,
{"status": "active"}
)
# Paginated queries
from commondao import Paged
result: Paged[UserModel] = await db.select_paged(
"select * from users where status = :status",
UserModel,
{"status": "active"},
size=10,
offset=0
)
print(f"Total users: {result.total}")
print(f"Current page: {len(result.items)} users")
```
### Raw SQL Execution
```python
# Execute a query and return results
rows = await db.execute_query(
"SELECT * FROM users WHERE created_at > :date",
{"date": "2023-01-01"}
)
# Execute a mutation and return affected row count
affected = await db.execute_mutation(
"UPDATE users SET status = :status WHERE last_login < :cutoff",
{"status": "inactive", "cutoff": "2023-01-01"}
)
```
### Transactions
```python
async with connect(host='localhost', user='root', db='testdb') as db:
# Start transaction (autocommit=False by default)
await db.insert('orders', data={'customer_id': 1, 'total': 99.99})
await db.insert('order_items', data={'order_id': db.lastrowid(), 'product_id': 42})
# Commit the transaction
await db.commit()
```
## 🔐 Type Safety
CommonDAO provides robust type checking to help prevent errors:
```python
from commondao import is_row_dict, is_query_dict
from typing import Dict, Any
# Valid row dict (for updates/inserts)
valid_data: Dict[str, Any] = {
"id": 1,
"name": "John",
"created_at": datetime.now(),
}
# Check type safety
assert is_row_dict(valid_data) # Type check passes
await db.update_by_key('users', key={'id': 1}, data=valid_data) # Type check passes
# Invalid row dict (contains a list)
invalid_data: Dict[str, Any] = {
"id": 1,
"tags": ["admin", "user"] # Lists are not valid row values
}
assert not is_row_dict(invalid_data) # Type check fails
```
## 📖 API Documentation
For complete API documentation, please see the docstrings in the code or visit our documentation website.
## 🧪 Testing
CommonDAO comes with comprehensive tests to ensure reliability:
```bash
# Install test dependencies
pip install -e ".[test]"
# Run tests
pytest tests
```
## 🤝 Contributing
Contributions are welcome! Please feel free to submit a Pull Request.
## 📄 License
This project is licensed under the Apache License 2.0.
Raw data
{
"_id": null,
"home_page": null,
"name": "commondao",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.10",
"maintainer_email": null,
"keywords": "mysql",
"author": null,
"author_email": "qorzj <goodhorsezxj@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/8e/0e/bf49e7864c220becc12de4fbf35898efb9278de603dac35d3f81a871538e/commondao-1.0.1.tar.gz",
"platform": null,
"description": "# CommonDAO\n\nA powerful, type-safe, and Pydantic-integrated async MySQL toolkit for Python.\n\n\n\n\n\n\nCommonDAO is a lightweight, type-safe async MySQL toolkit designed to simplify database operations with a clean, intuitive API. It integrates seamlessly with Pydantic for robust data validation while providing a comprehensive set of tools for common database tasks.\n\n## \u2728 Features\n\n- **Async/Await Support**: Built on aiomysql for non-blocking database operations\n- **Type Safety**: Strong typing with Python's type hints and runtime type checking\n- **Pydantic Integration**: Seamless validation and transformation of database records to Pydantic models\n- **SQL Injection Protection**: Parameterized queries for secure database access\n- **Comprehensive CRUD Operations**: Simple methods for common database tasks\n- **Raw SQL Support**: Full control when you need it with parameterized raw SQL\n- **Connection Pooling**: Efficient database connection management\n- **Minimal Boilerplate**: Write less code while maintaining readability and control\n\n## \ud83d\ude80 Installation\n\n```bash\npip install commondao\n```\n\n## \ud83d\udd0d Quick Start\n\n```python\nimport asyncio\nfrom commondao import connect\nfrom pydantic import BaseModel\n\n# Define your Pydantic model\nclass User(BaseModel):\n id: int\n name: str\n email: str\n\nasync def main():\n # Connect to database\n config = {\n 'host': 'localhost',\n 'port': 3306,\n 'user': 'root',\n 'password': 'password',\n 'db': 'testdb',\n 'autocommit': True,\n }\n \n async with connect(**config) as db:\n # Insert a new user\n await db.insert('users', data={'name': 'John Doe', 'email': 'john@example.com'})\n # Query the user with Pydantic model validation\n user = await db.select_one(\n \"select * from users where email = :email\",\n User,\n {\"email\": \"john@example.com\"}\n )\n print(f\"User: {user.name} ({user.email})\") # Output => User: John Doe (john@example.com)\n\nif __name__ == \"__main__\":\n asyncio.run(main())\n```\n\n## \ud83d\udcca Core Operations\n\n### Connection\n\n```python\nfrom commondao import connect\n\nasync with connect(\n host='localhost', \n port=3306, \n user='root', \n password='password', \n db='testdb'\n) as db:\n # Your database operations here\n pass\n```\n\n### Insert Data\n\n```python\n# Simple insert\nawait db.insert('users', data={'name': 'John', 'email': 'john@example.com'})\n\n# Insert with ignore option (skips duplicate key errors)\nawait db.insert('users', data={'name': 'Jane', 'email': 'jane@example.com'}, ignore=True)\n```\n\n### Update Data\n\n```python\n# Update where id = 1\nawait db.update_by_key(\n 'users', \n key={'id': 1}, \n data={'name': 'John Smith', 'email': 'john.smith@example.com'}\n)\n\n# Only non-None values will be updated\nawait db.update_by_key(\n 'users',\n key={'id': 1},\n data={'name': 'Jane Doe', 'email': None} # email won't be updated\n)\n```\n\n### Delete Data\n\n```python\n# Delete where id = 1\nawait db.delete_by_key('users', key={'id': 1})\n```\n\n### Query Data\n\n```python\n# Get a single row\nuser = await db.get_by_key('users', key={'id': 1})\n\n# Get a row or raise NotFoundError if not found\nuser = await db.get_by_key_or_fail('users', key={'id': 1})\n\n# Use with Pydantic models\nfrom pydantic import BaseModel\nfrom commondao import RawSql\nfrom typing import Annotated\n\nclass UserModel(BaseModel):\n id: int\n name: str\n email: str\n full_name: Annotated[str, RawSql(\"CONCAT(first_name, ' ', last_name)\")]\n\n# Query with model validation\nuser = await db.select_one(\n \"select * from users where id = :id\",\n UserModel,\n {\"id\": 1}\n)\n\n# Query multiple rows\nusers = await db.select_all(\n \"select * from users where status = :status\",\n UserModel,\n {\"status\": \"active\"}\n)\n\n# Paginated queries\nfrom commondao import Paged\n\nresult: Paged[UserModel] = await db.select_paged(\n \"select * from users where status = :status\",\n UserModel,\n {\"status\": \"active\"},\n size=10,\n offset=0\n)\n\nprint(f\"Total users: {result.total}\")\nprint(f\"Current page: {len(result.items)} users\")\n```\n\n### Raw SQL Execution\n\n```python\n# Execute a query and return results\nrows = await db.execute_query(\n \"SELECT * FROM users WHERE created_at > :date\",\n {\"date\": \"2023-01-01\"}\n)\n\n# Execute a mutation and return affected row count\naffected = await db.execute_mutation(\n \"UPDATE users SET status = :status WHERE last_login < :cutoff\",\n {\"status\": \"inactive\", \"cutoff\": \"2023-01-01\"}\n)\n```\n\n### Transactions\n\n```python\nasync with connect(host='localhost', user='root', db='testdb') as db:\n # Start transaction (autocommit=False by default)\n await db.insert('orders', data={'customer_id': 1, 'total': 99.99})\n await db.insert('order_items', data={'order_id': db.lastrowid(), 'product_id': 42})\n # Commit the transaction\n await db.commit()\n```\n\n## \ud83d\udd10 Type Safety\n\nCommonDAO provides robust type checking to help prevent errors:\n\n```python\nfrom commondao import is_row_dict, is_query_dict\nfrom typing import Dict, Any\n\n# Valid row dict (for updates/inserts)\nvalid_data: Dict[str, Any] = {\n \"id\": 1,\n \"name\": \"John\",\n \"created_at\": datetime.now(),\n}\n\n# Check type safety\nassert is_row_dict(valid_data) # Type check passes\nawait db.update_by_key('users', key={'id': 1}, data=valid_data) # Type check passes\n\n# Invalid row dict (contains a list)\ninvalid_data: Dict[str, Any] = {\n \"id\": 1,\n \"tags\": [\"admin\", \"user\"] # Lists are not valid row values\n}\n\nassert not is_row_dict(invalid_data) # Type check fails\n```\n\n## \ud83d\udcd6 API Documentation\n\nFor complete API documentation, please see the docstrings in the code or visit our documentation website.\n\n## \ud83e\uddea Testing\n\nCommonDAO comes with comprehensive tests to ensure reliability:\n\n```bash\n# Install test dependencies\npip install -e \".[test]\"\n\n# Run tests\npytest tests\n```\n\n## \ud83e\udd1d Contributing\n\nContributions are welcome! Please feel free to submit a Pull Request.\n\n## \ud83d\udcc4 License\n\nThis project is licensed under the Apache License 2.0.\n",
"bugtrack_url": null,
"license": "Apache-2.0",
"summary": "Mysql toolkit",
"version": "1.0.1",
"project_urls": {
"homepage": "https://github.com/lessweb/commondao"
},
"split_keywords": [
"mysql"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "7e9b2c3706f308220c8d7a42aa78b34d7e5d484ef9e6b842f650480b986fdc0a",
"md5": "274b903f52d49cebb2d7e89b3c1cfdc2",
"sha256": "a6d4dde7fdc11e2c858fff05b29f90e86736b51d66f962aee14ac318e39764a1"
},
"downloads": -1,
"filename": "commondao-1.0.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "274b903f52d49cebb2d7e89b3c1cfdc2",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.10",
"size": 16510,
"upload_time": "2025-07-23T04:10:38",
"upload_time_iso_8601": "2025-07-23T04:10:38.645543Z",
"url": "https://files.pythonhosted.org/packages/7e/9b/2c3706f308220c8d7a42aa78b34d7e5d484ef9e6b842f650480b986fdc0a/commondao-1.0.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "8e0ebf49e7864c220becc12de4fbf35898efb9278de603dac35d3f81a871538e",
"md5": "85e2122b2170c570aea3c3e137e51ca8",
"sha256": "c1bdbbf0a3d76ebb59857e0bff8371fb08f079ea32a27c6456d74c16685acf8a"
},
"downloads": -1,
"filename": "commondao-1.0.1.tar.gz",
"has_sig": false,
"md5_digest": "85e2122b2170c570aea3c3e137e51ca8",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.10",
"size": 29723,
"upload_time": "2025-07-23T04:10:40",
"upload_time_iso_8601": "2025-07-23T04:10:40.101586Z",
"url": "https://files.pythonhosted.org/packages/8e/0e/bf49e7864c220becc12de4fbf35898efb9278de603dac35d3f81a871538e/commondao-1.0.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-07-23 04:10:40",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "lessweb",
"github_project": "commondao",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "commondao"
}
