Banco de Dados

O Velox inclui um ORM completo com suporte a SQLite e PostgreSQL.

Configuração

Escolha o banco de dados via variável de ambiente:

# SQLite (padrão)
DATABASE_URI=db/app.db

# PostgreSQL
DATABASE_URI=postgresql://user:pass@localhost:5432/mydb

Criando umaConexão

from velox.database import Database

# SQLite (padrão)
db = Database('db/app.db')

# PostgreSQL
db = Database('postgresql://user:pass@localhost:5432/mydb')

# Use contexto
with db:
    rows = db.fetchall('SELECT * FROM users')

Operações Básicas

from velox.database import Database

db = Database('db/app.db')

# INSERT
db.execute('INSERT INTO users (name) VALUES (?)', ('João',))

# SELECT one
user = db.fetchone('SELECT * FROM users WHERE id = ?', (1,))

# SELECT all
users = db.fetchall('SELECT * FROM users')

Modelo (ORM)

from velox.database import Model

class User(Model):
    table = 'users'
    schema = {
        'name': str,
        'email': str,
        'is_active': bool,
    }

# Criar tabela
User.create_table()

# CREATE
user = User.create(name='João', email='joao@test.com')

# READ
user = User.get(1)
all_users = User.all()

# UPDATE
user.update(name='João atualizado')

# DELETE
user.delete()

Queries Encadeadas

from velox.database import Model

class Post(Model):
    table = 'posts'
    schema = {'title': str, 'content': str, 'published': bool}

#WHERE + ORDER + LIMIT
posts = (Post
    .where('published', '=', True)
    .order_by('created_at', 'DESC')
    .limit(10)
    .get())

# Paginação
result = Post.paginate(page=1, per_page=20)

# Contagem
count = Post.count(published=True)
exists = Post.exists(email='joao@test.com')

Relacionamentos

from velox.database import Model, ForeignKey, ManyToMany

class Author(Model):
    table = 'authors'
    schema = {'name': str}
    books = ManyToMany('Book', through='author_books')

class Book(Model):
    table = 'books'
    schema = {'title': str}
    author = ForeignKey('Author', on_delete='CASCADE')

# Acessar relacionados
author = Author.get(1)
books = author.related('books')

PostgreSQL

Instale o driver:

pip install psycopg2-binary

Configuração:

import os
os.environ['DATABASE_URI'] = 'postgresql://user:pass@localhost:5432/mydb'

Async Database

Para modo ASGI, use banco assíncrono:

from velox.database import AsyncDatabase

db = AsyncDatabase('db/app.db')

# PostgreSQL
db = AsyncDatabase('postgresql://user:pass@localhost:5432/mydb')

async def handler(req, res):
    rows = await db.fetchall('SELECT * FROM posts')
    res.json(rows)

Instale os drivers:

pip install aiosqlite     # SQLite async
pip install asyncpg       # PostgreSQL async

Próximos Passos

  • Leia models para modelos avançados

  • Explore queries para queries complexas