FireO – ORM для Google Firestore

Google Firestore — это облачная NoSQL база данных от Firebase, предназначенная для хранения JSON-документов. Она отлично подходит для масштабируемых, real-time приложений. Однако работать с Firestore напрямую через google-cloud-firestore не всегда удобно, особенно если вы привыкли к объектно-ориентированному подходу и типизированному коду.

FireO – ORM для Google Firestore — это Python-библиотека, которая упрощает работу с Firestore, предоставляя интерфейс, напоминающий популярные ORM. Модели, поля, валидация, связи — всё это теперь доступно и в мире Firebase.

Что такое FireO и почему стоит его использовать

FireO (Firebase ORM) — это объектно-реляционное отображение для Google Firestore, написанное на Python. Библиотека позволяет разработчикам работать с Firestore как с традиционной реляционной базой данных, используя классы и объекты вместо прямых запросов к API.

Основные преимущества FireO:

• Простота использования: Интуитивно понятный синтаксис, похожий на Django ORM
• Типизация полей: Строгая типизация с автоматической валидацией
• Валидация данных: Встроенная проверка данных перед сохранением
• Поддержка связей: Работа с reference полями и вложенными структурами
• Автоматическое управление метаданными: Автоматическое создание временных меток
• Миграции: Упрощенное управление структурой данных

Установка и настройка FireO

Установка библиотеки

pip install fireo

Настройка аутентификации

Для работы с Firestore необходимо настроить аутентификацию через Service Account:

1. Создайте Service Account в Firebase Console
2. Скачайте JSON-ключ
3. Укажите путь к ключу в переменной окружения:

export GOOGLE_APPLICATION_CREDENTIALS="path/to/your-key.json"

Альтернативная настройка подключения

from fireo import connection

# Подключение через файл
connection(from_file="path/to/serviceAccountKey.json")

# Подключение через переменную окружения
connection()

Основы Firestore и документо-ориентированной модели

Структура данных в Firestore

Firestore хранит данные в следующей иерархии:

• База данных → Коллекции → Документы → Поля
• Документы могут содержать подколлекции
• Поля могут быть различных типов: строки, числа, массивы, объекты и т.д.

Особенности NoSQL подхода

Firestore является безсхемной базой данных, что означает:

• Документы в одной коллекции могут иметь разные поля
• Нет необходимости заранее определять структуру
• Гибкость в изменении структуры данных

FireO добавляет уровень структуры и валидации поверх этой гибкости.

Создание моделей в FireO

Базовая модель

from fireo.models import Model
from fireo.fields import TextField, NumberField, BooleanField

class User(Model):
    name = TextField(required=True, max_length=100)
    age = NumberField(required=True)
    is_active = BooleanField(default=True)
    
    class Meta:
        collection_name = "users"  # Опционально, по умолчанию используется название класса

Расширенная модель с валидацией

from fireo.fields import EmailField, DateTimeField, ListField
from datetime import datetime

class Profile(Model):
    user_id = TextField(required=True)
    email = EmailField(required=True)
    created_at = DateTimeField(default=datetime.now)
    tags = ListField()
    preferences = MapField()
    
    class Meta:
        collection_name = "profiles"
        to_lowercase = True  # Автоматическое преобразование в нижний регистр

Подробное описание полей FireO

Текстовые поля

# Базовое текстовое поле
name = TextField(required=True, max_length=50)

# Поле с выбором из вариантов
status = TextField(choices=["active", "inactive", "pending"])

# Поле с значением по умолчанию
description = TextField(default="No description")

Числовые поля

# Целые числа
age = NumberField(required=True)

# Числа с ограничениями
price = NumberField(min_value=0, max_value=999999)

# Числа с значением по умолчанию
rating = NumberField(default=0)

Специальные поля

# Логическое поле
is_published = BooleanField(default=False)

# Поле даты и времени
created_at = DateTimeField(auto=True)
updated_at = DateTimeField(auto_add=True)

# Поле для списков
tags = ListField(default=[])

# Поле для вложенных объектов
metadata = MapField(default={})

CRUD-операции: создание, чтение, обновление, удаление

Создание документов

# Создание с автоматическим ID
user = User(name="Иван Петров", age=30)
user.save()
print(f"Создан пользователь с ID: {user.key}")

# Создание с заданным ID
user = User(name="Анна Сидорова", age=25)
user.key = "custom_user_id"
user.save()

Чтение данных

# Получение по ID
user = User.collection.get("user_id")

# Получение с обработкой ошибок
try:
    user = User.collection.get("non_existent_id")
except Exception as e:
    print(f"Пользователь не найден: {e}")

# Получение всех документов
all_users = User.collection.fetch()

Обновление документов

# Обновление существующего документа
user = User.collection.get("user_id")
user.name = "Новое имя"
user.age = 31
user.update()

# Частичное обновление
user.update(["name"])  # Обновить только поле name

Удаление документов

# Удаление документа
user = User.collection.get("user_id")
user.delete()

# Удаление по условию (требует дополнительной логики)
users_to_delete = User.collection.filter("is_active", "==", False).fetch()
for user in users_to_delete:
    user.delete()

Запросы и фильтрация данных

Базовая фильтрация

# Простые фильтры
active_users = User.collection.filter("is_active", "==", True).fetch()
adult_users = User.collection.filter("age", ">=", 18).fetch()

# Фильтр по массиву
users_with_tag = User.collection.filter("tags", "array_contains", "premium").fetch()

Сложные запросы

# Множественные фильтры
young_active_users = User.collection.filter("age", "<", 30).filter("is_active", "==", True).fetch()

# Сортировка
sorted_users = User.collection.order("age").fetch()
reverse_sorted = User.collection.order("age", direction="desc").fetch()

# Ограничение результатов
first_10_users = User.collection.limit(10).fetch()

Пагинация

# Постраничная навигация
page_size = 10
users_page1 = User.collection.limit(page_size).fetch()

# Для следующей страницы используется cursor
last_user = users_page1[-1]
users_page2 = User.collection.start_after(last_user).limit(page_size).fetch()

Работа с отношениями и связями

ReferenceField для связей между документами

from fireo.fields import ReferenceField

class Post(Model):
    title = TextField(required=True)
    content = TextField()
    author = ReferenceField(User, required=True)
    category = ReferenceField("Category")  # Можно указать строкой
    
    class Meta:
        collection_name = "posts"

# Создание связанного документа
user = User.collection.get("user_id")
post = Post(title="Мой пост", content="Содержимое", author=user)
post.save()

# Получение связанных данных
post = Post.collection.get("post_id")
author = post.author.get()  # Получить связанного пользователя

Вложенные модели с EmbeddedField

from fireo.fields import EmbeddedField

class Address(Model):
    street = TextField()
    city = TextField()
    country = TextField()
    
    class Meta:
        ignore_none_field = False

class Company(Model):
    name = TextField()
    address = EmbeddedField(Address)
    
# Использование
address = Address(street="Пушкинская 1", city="Москва", country="Россия")
company = Company(name="Моя компания", address=address)
company.save()

Продвинутые возможности валидации

Кастомные валидаторы

from fireo.fields import TextField
from fireo.models import Model

def validate_phone(value):
    if not value.startswith('+'):
        raise ValueError("Телефон должен начинаться с +")
    return value

class Contact(Model):
    name = TextField(required=True)
    phone = TextField(validator=validate_phone)
    
    def clean(self):
        # Кастомная валидация на уровне модели
        if self.name and len(self.name) < 2:
            raise ValueError("Имя должно содержать минимум 2 символа")

Автоматические поля

from fireo.fields import DateTimeField, IDField
from datetime import datetime

class Document(Model):
    id = IDField()  # Автоматический ID
    title = TextField()
    created_at = DateTimeField(auto=True)  # Автоматическая установка при создании
    updated_at = DateTimeField(auto_add=True)  # Обновляется при каждом сохранении
    
    class Meta:
        ignore_none_field = True  # Игнорировать поля со значением None

Работа с транзакциями

Базовые транзакции

from fireo import transaction

@transaction
def transfer_points(from_user_id, to_user_id, points):
    from_user = User.collection.get(from_user_id)
    to_user = User.collection.get(to_user_id)
    
    if from_user.points < points:
        raise ValueError("Недостаточно баллов")
    
    from_user.points -= points
    to_user.points += points
    
    from_user.update()
    to_user.update()
    
    return True

# Использование
try:
    transfer_points("user1", "user2", 100)
    print("Перевод выполнен успешно")
except Exception as e:
    print(f"Ошибка перевода: {e}")

Интеграция с веб-фреймворками

Интеграция с FastAPI

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional

app = FastAPI()

class UserCreate(BaseModel):
    name: str
    age: int
    email: str

class UserResponse(BaseModel):
    id: str
    name: str
    age: int
    email: str

@app.post("/users", response_model=UserResponse)
async def create_user(user_data: UserCreate):
    user = User(
        name=user_data.name,
        age=user_data.age,
        email=user_data.email
    )
    user.save()
    
    return UserResponse(
        id=user.key,
        name=user.name,
        age=user.age,
        email=user.email
    )

@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: str):
    try:
        user = User.collection.get(user_id)
        return UserResponse(
            id=user.key,
            name=user.name,
            age=user.age,
            email=user.email
        )
    except Exception:
        raise HTTPException(status_code=404, detail="Пользователь не найден")

@app.get("/users", response_model=List[UserResponse])
async def list_users(limit: int = 10, offset: int = 0):
    users = User.collection.limit(limit).offset(offset).fetch()
    return [
        UserResponse(
            id=user.key,
            name=user.name,
            age=user.age,
            email=user.email
        )
        for user in users
    ]

Интеграция с Flask

from flask import Flask, request, jsonify
from werkzeug.exceptions import NotFound

app = Flask(__name__)

@app.route('/users', methods=['POST'])
def create_user():
    data = request.json
    user = User(
        name=data['name'],
        age=data['age'],
        email=data.get('email')
    )
    user.save()
    return jsonify(user.to_dict()), 201

@app.route('/users/<user_id>', methods=['GET'])
def get_user(user_id):
    try:
        user = User.collection.get(user_id)
        return jsonify(user.to_dict())
    except Exception:
        raise NotFound("Пользователь не найден")

@app.route('/users/<user_id>', methods=['PUT'])
def update_user(user_id):
    data = request.json
    try:
        user = User.collection.get(user_id)
        user.name = data.get('name', user.name)
        user.age = data.get('age', user.age)
        user.update()
        return jsonify(user.to_dict())
    except Exception:
        raise NotFound("Пользователь не найден")

@app.route('/users/<user_id>', methods=['DELETE'])
def delete_user(user_id):
    try:
        user = User.collection.get(user_id)
        user.delete()
        return '', 204
    except Exception:
        raise NotFound("Пользователь не найден")

Тестирование приложений с FireO

Настройка тестовой среды

import os
import pytest
from fireo import connection

@pytest.fixture(scope="session")
def firestore_emulator():
    # Настройка эмулятора Firestore
    os.environ["FIRESTORE_EMULATOR_HOST"] = "localhost:8080"
    connection()
    yield
    # Очистка после тестов

@pytest.fixture
def clean_firestore():
    # Очистка коллекций перед каждым тестом
    yield
    # Логика очистки

Примеры тестов

def test_user_creation(firestore_emulator, clean_firestore):
    user = User(name="Test User", age=25)
    user.save()
    
    assert user.key is not None
    assert user.name == "Test User"
    assert user.age == 25

def test_user_retrieval(firestore_emulator, clean_firestore):
    # Создаем пользователя
    user = User(name="Test User", age=25)
    user.save()
    user_id = user.key
    
    # Получаем пользователя
    retrieved_user = User.collection.get(user_id)
    
    assert retrieved_user.name == "Test User"
    assert retrieved_user.age == 25

def test_user_filtering(firestore_emulator, clean_firestore):
    # Создаем несколько пользователей
    User(name="Alice", age=20).save()
    User(name="Bob", age=30).save()
    User(name="Charlie", age=25).save()
    
    # Фильтрация
    young_users = User.collection.filter("age", "<", 25).fetch()
    assert len(young_users) == 1
    assert young_users[0].name == "Alice"

Производительность и оптимизация

Оптимизация запросов

# Плохо: множественные запросы
users = User.collection.fetch()
for user in users:
    posts = Post.collection.filter("author", "==", user.key).fetch()
    print(f"{user.name} имеет {len(posts)} постов")

# Хорошо: batch запросы
user_ids = [user.key for user in users]
posts = Post.collection.filter("author", "in", user_ids).fetch()
posts_by_user = {}
for post in posts:
    if post.author not in posts_by_user:
        posts_by_user[post.author] = []
    posts_by_user[post.author].append(post)

Кеширование

from functools import lru_cache

@lru_cache(maxsize=100)
def get_user_cached(user_id):
    return User.collection.get(user_id)

# Использование кеша на уровне приложения
import time

class UserCache:
    def __init__(self, ttl=300):  # 5 минут
        self.cache = {}
        self.ttl = ttl
    
    def get(self, user_id):
        now = time.time()
        if user_id in self.cache:
            user, timestamp = self.cache[user_id]
            if now - timestamp < self.ttl:
                return user
        
        user = User.collection.get(user_id)
        self.cache[user_id] = (user, now)
        return user

Миграции и управление схемой

Создание миграций

class Migration001:
    """Добавление поля email к пользователям"""
    
    def up(self):
        users = User.collection.fetch()
        for user in users:
            if not hasattr(user, 'email'):
                user.email = f"{user.name.lower().replace(' ', '.')}@example.com"
                user.update()
    
    def down(self):
        # Откат миграции
        users = User.collection.fetch()
        for user in users:
            if hasattr(user, 'email'):
                delattr(user, 'email')
                user.update()

Полная таблица методов и функций FireO

Обработка ошибок и исключений

Основные исключения FireO

from fireo.exceptions import ValidationError, NotFound, PermissionDenied

try:
    user = User(name="", age=-5)  # Невалидные данные
    user.save()
except ValidationError as e:
    print(f"Ошибка валидации: {e}")

try:
    user = User.collection.get("non_existent_id")
except NotFound as e:
    print(f"Документ не найден: {e}")

try:
    user = User.collection.get("restricted_id")
except PermissionDenied as e:
    print(f"Доступ запрещен: {e}")

Глобальная обработка ошибок

import logging

def handle_firestore_error(func):
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except ValidationError as e:
            logging.error(f"Validation error in {func.__name__}: {e}")
            raise
        except NotFound as e:
            logging.error(f"Document not found in {func.__name__}: {e}")
            raise
        except Exception as e:
            logging.error(f"Unexpected error in {func.__name__}: {e}")
            raise
    return wrapper

@handle_firestore_error
def create_user(name, age):
    user = User(name=name, age=age)
    user.save()
    return user

Мониторинг и логирование

Настройка логирования

import logging
from fireo import connection

# Настройка логирования FireO
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('fireo')

# Логирование операций
class LoggedUser(User):
    def save(self):
        logger.info(f"Сохранение пользователя: {self.name}")
        super().save()
        logger.info(f"Пользователь сохранен с ID: {self.key}")
    
    def delete(self):
        logger.info(f"Удаление пользователя: {self.key}")
        super().delete()
        logger.info(f"Пользователь удален")

Метрики производительности

import time
from contextlib import contextmanager

@contextmanager
def measure_time(operation_name):
    start_time = time.time()
    yield
    end_time = time.time()
    print(f"{operation_name} выполнено за {end_time - start_time:.2f} секунд")

# Использование
with measure_time("Создание пользователя"):
    user = User(name="Test", age=25)
    user.save()

with measure_time("Запрос пользователей"):
    users = User.collection.limit(100).fetch()

Сравнение с другими библиотеками

Часто задаваемые вопросы

Что такое FireO и чем он отличается от прямой работы с Firestore?

FireO — это ORM (Object-Relational Mapping) для Google Firestore, который предоставляет объектно-ориентированный интерфейс для работы с документами. В отличие от прямой работы с Firestore API, FireO предлагает типизацию полей, автоматическую валидацию, простой синтаксис для создания моделей и встроенную поддержку связей между документами.

Поддерживает ли FireO асинхронные операции?

На данный момент FireO работает только синхронно. Для асинхронных операций можно использовать обёртки через asyncio.run_in_executor() или переключиться на google-cloud-firestore с нативной поддержкой async/await.

Как работают транзакции в FireO?

FireO поддерживает транзакции через декоратор @transaction. Все операции внутри транзакции выполняются атомарно — либо все успешно, либо все откатываются в случае ошибки.

Можно ли использовать FireO с вложенными коллекциями?

Да, FireO поддерживает работу с подколлекциями через метод parent(). Например: Order.collection.parent("users/user_id").create(data).

Как тестировать приложения с FireO?

Рекомендуется использовать Firebase Emulator для локального тестирования. Настройте переменную окружения FIRESTORE_EMULATOR_HOST=localhost:8080 и используйте эмулятор для изолированного тестирования.

Поддерживает ли FireO миграции схемы данных?

FireO не имеет встроенной системы миграций, но вы можете создавать собственные скрипты миграций для обновления структуры данных в существующих документах.

Какие ограничения есть у FireO по сравнению с нативным Firestore API?

FireO может иметь ограничения в сложных запросах, некоторых специфичных для Firestore операциях и производительности при больших объемах данных. Для максимальной гибкости рекомендуется комбинировать FireO с прямым использованием Firestore API.

Можно ли использовать FireO в продакшене?

Да, FireO подходит для продакшена, особенно для малых и средних проектов. Однако для крупных высоконагруженных приложений рекомендуется тщательное тестирование производительности и возможно использование нативного Firestore API для критичных операций.

Заключение

FireO представляет собой мощную библиотеку для работы с Google Firestore в Python-приложениях. Она значительно упрощает разработку, предоставляя знакомый ORM-подобный интерфейс, строгую типизацию и автоматическую валидацию данных.

Основные преимущества FireO включают простоту использования, хорошую интеграцию с популярными веб-фреймворками и достаточную функциональность для большинства задач. Однако для проектов, требующих максимальной производительности или сложных асинхронных операций, стоит рассмотреть альтернативные решения.

FireO отлично подходит для:

• Быстрого прототипирования
• Малых и средних веб-приложений
• Проектов, где важна читаемость и поддерживаемость кода
• Команд, знакомых с Django ORM или подобными решениями

Независимо от выбора, FireO остается одним из лучших инструментов для работы с Firestore в экосистеме Python.