piccolo

Piccolo is an async ORM and query builder for Python, designed to work seamlessly with PostgreSQL. It offers a simple and intuitive API for database operations, supports migrations, and provides a powerful admin interface for managing data.

Pros

• Asynchronous by default, allowing for efficient database operations
• Intuitive query syntax that closely resembles Python code
• Built-in support for database migrations and schema management
• Includes a customizable admin interface for easy data management

Cons

• Primarily focused on PostgreSQL, with limited support for other databases
• Relatively new compared to more established ORMs like SQLAlchemy
• Learning curve for developers not familiar with async programming
• Documentation could be more comprehensive for advanced use cases

Code Examples

1. Defining a table:

from piccolo.table import Table
from piccolo.columns import Varchar, Integer

class User(Table):
    name = Varchar()
    age = Integer()

2. Performing a query:

# Fetch all users over 18
users = await User.select().where(User.age > 18)

# Print user names
for user in users:
    print(user.name)

3. Inserting data:

# Create a new user
new_user = User(name="Alice", age=25)
await new_user.save()

4. Updating data:

# Update all users' age by 1
await User.update({User.age: User.age + 1}).run()

Getting Started

1. Install Piccolo:

pip install piccolo

2. Create a database configuration:

# config.py
from piccolo.engine.postgres import PostgresEngine

DB = PostgresEngine(
    config={
        "database": "my_db",
        "user": "postgres",
        "password": "password",
        "host": "localhost",
        "port": 5432,
    }
)

3. Define your tables and run migrations:

# tables.py
from piccolo.table import Table
from piccolo.columns import Varchar, Integer

class User(Table):
    name = Varchar()
    age = Integer()

# Run in terminal:
# piccolo migrations new
# piccolo migrations run

4. Use Piccolo in your application:

from tables import User

async def main():
    await User.create_table().run()
    
    new_user = User(name="Bob", age=30)
    await new_user.save()
    
    users = await User.select().where(User.age > 25)
    for user in users:
        print(f"{user.name} is {user.age} years old")

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