cocobase-framework 0.1.0

Self-hosted Backend as a Service with real-time, events, and SQL power

🚀 CocoBase v2.0 - Self-Hosted Backend as a Service

A production-ready, self-hosted Backend as a Service framework with real-time capabilities, event system, file uploads, and SQL power. Built with FastAPI and SQLModel for modern Python applications.

✨ What Makes CocoBase Special

• 🗄️ Real SQL Tables - Traditional database architecture (not JSON storage)
• 🎯 Event-Driven - 13 lifecycle hooks for custom business logic
• 📡 Real-time - WebSocket subscriptions with filtering
• 📁 File Storage - Built-in with cloud provider support (S3, Cloudinary, B2)
• 💻 SQL Editor - Execute queries from the dashboard
• 🔌 Extensible - Add custom routes with full database access
• 👤 Built-in Auth - User model with bcrypt password hashing
• 📊 Zero Config - Works immediately out of the box
• 🎨 Web Dashboard - Visual interface for everything
• 📖 Auto Docs - Interactive API documentation

---

🎉 New in v2.0

1. Event System

Register callbacks for lifecycle events:

@cocobase.events.on(EventType.AFTER_CREATE, collection="users")
async def on_user_created(ctx: EventContext):
    print(f"New user: {ctx.data['username']}")

2. Real-time WebSocket

Watch collections for live updates:

ws.send(
  JSON.stringify({
    action: "subscribe",
    collection: "users",
    filters: { is_admin: true },
  })
);

3. SQL Editor

Execute raw SQL from the dashboard with query history!

4. File Uploads

Built-in file handling with local/cloud storage.

5. Database Access

Use the database directly in custom routes:

@app.get("/custom")
async def my_route(session: Session = Depends(cocobase.get_db)):
    return session.exec(text("SELECT * FROM users"))

---

📦 Installation

pip install cocobase

Or from source:

git clone https://github.com/lordace-coder/cocobase_framework.git
cd cocobase
pip install -e .

---

⚡ Quick Start

1. Create app.py

from cocobase import CocoBase

cocobase = CocoBase()
app = cocobase.app

if __name__ == "__main__":
    cocobase.run()

2. Run

python app.py

3. Access

• Dashboard: http://localhost:8000
• API Docs: http://localhost:8000/docs
• SQL Editor: Dashboard → SQL Editor tab

---

🎯 Core Features

Traditional Database Tables

Each collection is a real SQL table with proper columns and types:

# Create a collection via API
{
  "name": "products",
  "columns": [
    {"name": "id", "type": "integer", "primary_key": true, "auto_increment": true},
    {"name": "name", "type": "string", "required": true},
    {"name": "price", "type": "float"},
    {"name": "in_stock", "type": "boolean", "default": true}
  ]
}

Supported Types: string, integer, float, boolean, datetime, text, json, file, files

Built-in User Model

Ready-to-use authentication with password hashing:

# Create user (password auto-hashed)
POST /collections/users/documents
{
  "username": "john",
  "email": "[e-mail]",
  "password": "secret123"
}

Event-Driven Architecture

Hook into document lifecycle:

from cocobase import EventType, EventContext

# Validate before save
@cocobase.events.on(EventType.BEFORE_CREATE, collection="users")
async def validate_user(ctx: EventContext):
    if not ctx.data.get('email'):
        ctx.cancel()  # Prevent creation

# Send notification after save
@cocobase.events.on(EventType.AFTER_CREATE, collection="orders")
async def send_confirmation(ctx: EventContext):
    # Send email, update inventory, etc.
    pass

Available Events:

• APP_START, APP_SHUTDOWN
• BEFORE_CREATE, AFTER_CREATE
• BEFORE_UPDATE, AFTER_UPDATE
• BEFORE_DELETE, AFTER_DELETE
• COLLECTION_CREATED, COLLECTION_DELETED
• FILE_UPLOADED, FILE_DELETED

Real-time Subscriptions

Watch collections with WebSocket:

const ws = new WebSocket("ws://localhost:8000/ws");

// Subscribe to collection
ws.send(
  JSON.stringify({
    action: "subscribe",
    collection: "users",
    filters: {
      age_gte: 18, // Age >= 18
      status: "active",
    },
    events: ["create", "update", "delete"],
  })
);

// Receive updates
ws.onmessage = (event) => {
  const message = JSON.parse(event.data);
  if (message.type === "event") {
    console.log("Update:", message.data);
  }
};

Filter Operators: _eq, _ne, _gt, _gte, _lt, _lte, _contains, _in

File Upload System

Upload files with automatic storage:

# Upload endpoint
POST /files/upload
Content-Type: multipart/form-data

# Collection with file field
{
  "name": "posts",
  "columns": [
    {"name": "id", "type": "integer", "primary_key": true},
    {"name": "title", "type": "string"},
    {"name": "featured_image", "type": "file"},
    {"name": "attachments", "type": "files"}  // Multiple files
  ]
}

Access files at /uploads/filename.jpg

Custom Routes with Database Access

Extend CocoBase with custom endpoints:

from sqlmodel import Session, text
from fastapi import Depends

@app.get("/api/stats")
async def get_stats(session: Session = Depends(cocobase.get_db)):
    """Custom endpoint with direct database access"""
    result = session.execute(text("""
        SELECT COUNT(*) as total_users,
               SUM(CASE WHEN is_admin THEN 1 ELSE 0 END) as admins
        FROM users
    """))
    row = result.first()
    return {"total_users": row[0], "admins": row[1]}

---

📚 API Endpoints

Collections

GET    /collections                      # List all collections
POST   /collections                      # Create collection
DELETE /collections/{name}               # Delete collection
PATCH  /collections/{name}/add-column    # Add column dynamically

Documents

GET    /collections/{name}/documents           # List documents
POST   /collections/{name}/documents           # Create document
GET    /collections/{name}/documents/{id}      # Get document
PATCH  /collections/{name}/documents/{id}      # Update document
DELETE /collections/{name}/documents/{id}      # Delete document

Files

POST   /files/upload                     # Upload file
POST   /files/upload-multiple            # Upload multiple files
GET    /files/{filename}                 # Get file
DELETE /files/{filename}                 # Delete file

Real-time

WS     /ws                              # WebSocket endpoint
GET    /realtime/stats                  # Connection statistics

SQL

POST   /sql/execute                     # Execute SQL query

---

💡 Complete Examples

Example 1: Blog with Events

from cocobase import CocoBase, EventType, EventContext

cocobase = CocoBase()
app = cocobase.app

# Auto-generate slug from title
@cocobase.events.on(EventType.BEFORE_CREATE, collection="posts")
async def generate_slug(ctx: EventContext):
    title = ctx.data.get('title', '')
    ctx.data['slug'] = title.lower().replace(' ', '-')
    ctx.data['view_count'] = 0

# Notify followers
@cocobase.events.on(EventType.AFTER_CREATE, collection="posts")
async def notify_followers(ctx: EventContext):
    # Send notifications to followers
    print(f"New post: {ctx.data['title']}")

# Custom view counter
@app.post("/api/posts/{post_id}/view")
async def increment_views(post_id: int, session: Session = Depends(cocobase.get_db)):
    query = text("UPDATE posts SET view_count = view_count + 1 WHERE id = :id")
    session.execute(query, {"id": post_id})
    session.commit()
    return {"message": "View counted"}

if __name__ == "__main__":
    cocobase.run()

Example 2: Real-time Dashboard

<!DOCTYPE html>
<html>
  <head>
    <title>Live Dashboard</title>
  </head>
  <body>
    <h1>Live User Activity</h1>
    <div id="activity"></div>

    <script>
      const ws = new WebSocket("ws://localhost:8000/ws");

      ws.onopen = () => {
        ws.send(
          JSON.stringify({
            action: "subscribe",
            collection: "users",
            events: ["create", "update"],
          })
        );
      };

      ws.onmessage = (event) => {
        const msg = JSON.parse(event.data);
        if (msg.type === "event") {
          document
            .getElementById("activity")
            .insertAdjacentHTML(
              "afterbegin",
              `<div>${msg.event} in ${msg.collection}: ${JSON.stringify(
                msg.data
              )}</div>`
            );
        }
      };
    </script>
  </body>
</html>

Example 3: Data Validation

@cocobase.events.on(EventType.BEFORE_CREATE, collection="users")
async def validate_user(ctx: EventContext):
    email = ctx.data.get('email', '')

    # Validate email
    if '@' not in email:
        ctx.cancel()
        return

    # Check for duplicates
    with Session(cocobase.engine) as session:
        query = text("SELECT COUNT(*) FROM users WHERE email = :email")
        result = session.execute(query, {"email": email})
        if result.scalar() > 0:
            ctx.cancel()
            print(f"Email already exists: {email}")

---

🗄️ Database Support

# SQLite (default)
CocoBase(database_url="sqlite:///myapp.db")

# PostgreSQL
CocoBase(database_url="postgresql://user:pass@localhost/mydb")

# MySQL
CocoBase(database_url="mysql+pymysql://user:pass@localhost/mydb")

---

🔧 Configuration

cocobase = CocoBase(
    database_url="sqlite:///myapp.db",  # Database connection
    dev_mode=True,                      # Development mode
    enable_dashboard=True,              # Enable web dashboard
    upload_dir="uploads"                # File upload directory
)

---

📖 Documentation

• Quick Reference: QUICK_REFERENCE.md
• Developer Guide: docs/DEVELOPER_GUIDE.md
• Relationships & Files: docs/RELATIONSHIPS_AND_FILES.md
• Features Summary: FEATURES_SUMMARY.md
• Example App: example_usage.py

---

🎓 Learning Resources

1. Start Here: Read QUICK_REFERENCE.md
2. Run Example: python example_usage.py
3. Deep Dive: Read docs/DEVELOPER_GUIDE.md
4. Build Your App: Use the patterns and extend!

---

🏗️ Architecture

┌─────────────────────────────────────────────┐
│              CocoBase v2.0                  │
├─────────────────────────────────────────────┤
│  FastAPI + SQLModel + WebSocket + Events   │
│                                             │
│  ┌─────────┐  ┌─────────┐  ┌──────────┐   │
│  │ Events  │  │Real-time│  │ Storage  │   │
│  │ System  │  │WebSocket│  │ Service  │   │
│  └─────────┘  └─────────┘  └──────────┘   │
│                                             │
│  ┌───────────────────────────────────────┐ │
│  │   Real SQL Tables (Not JSON)          │ │
│  │   users | posts | products | ...     │ │
│  └───────────────────────────────────────┘ │
└─────────────────────────────────────────────┘

---

🔐 Security

⚠️ For Production:

1. Add authentication (JWT/OAuth2)
2. Enable HTTPS
3. Configure CORS properly
4. Add rate limiting
5. Validate all inputs
6. Use strong passwords
7. Keep dependencies updated

---

🚀 Deployment

Using Uvicorn

uvicorn app:app --host 0.0.0.0 --port 8000 --workers 4

Using Docker

FROM python:3.12
WORKDIR /app
COPY . .
RUN pip install -r requirements.txt
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

---

🤝 Contributing

Contributions welcome! Areas to improve:

• Additional storage providers (AWS S3, Azure Blob, etc.)
• Authentication providers (OAuth2, JWT, etc.)
• More event types
• Performance optimizations
• UI enhancements

---

📄 License

MIT License - Free for personal and commercial use!

---

🎉 Summary

CocoBase v2.0 gives you:

✅ Traditional SQL database (not JSON storage)
✅ Event-driven architecture (13 lifecycle hooks)
✅ Real-time WebSocket subscriptions
✅ File upload & storage
✅ SQL editor in dashboard
✅ Custom routes with database access
✅ Built-in authentication
✅ Zero configuration
✅ Production ready

Build your backend in minutes, not days! 🚀

---

🔗 Links

• GitHub: github.com/yourusername/cocobase
• Documentation: Read the Docs
• PyPI: pypi.org/project/cocobase

---

Built with ❤️ using FastAPI, SQLModel, and WebSockets