TinyDB – NoSQL база данных

Полное руководство по TinyDB: лёгкая NoSQL база данных для Python

Введение

Современные приложения не всегда нуждаются в мощных и сложных СУБД. В случаях, когда проект не требует масштабируемости или высокой скорости обработки большого объёма данных, достаточно лёгкой и встроенной базы. TinyDB – NoSQL база данных, полностью написанная на Python, представляет собой удобное и простое решение для хранения структурированных данных без SQL.

TinyDB сохраняет данные в формате JSON и не требует установки сервера. Это делает её идеальной для скриптов, небольших CLI-инструментов, десктопных приложений, прототипов и образовательных проектов. Библиотека весит всего несколько килобайт и имеет минимальные зависимости, что делает её отличным выбором для лёгких решений.

Установка и базовая настройка

Установка TinyDB

Установить TinyDB можно одной строкой через pip:

pip install tinydb

Для использования дополнительных возможностей доступны расширения:

pip install tinydb[compression]  # Для сжатия данных
pip install tinydb[yaml]         # Для работы с YAML

Создание базы данных

Создание базы с сохранением в файл:

from tinydb import TinyDB
db = TinyDB('db.json')  # Файл будет создан автоматически

Создание базы в памяти:

from tinydb.storages import MemoryStorage
db = TinyDB(storage=MemoryStorage)

Настройка кодировки и параметров

from tinydb import TinyDB
from tinydb.storages import JSONStorage

# Настройка кодировки
db = TinyDB('db.json', encoding='utf-8')

# Настройка отступов JSON для читаемости
db = TinyDB('db.json', indent=4, separators=(',', ': '))

Основы структуры данных TinyDB

TinyDB использует документо-ориентированную модель данных:

• Данные хранятся в виде списка документов (словари Python)
• Каждый документ имеет уникальный doc_id, автоматически генерируемый
• Поддерживаются типы: строки, числа, списки, словари, булевы значения и null
• Документы могут иметь различную структуру в одной таблице
• Поддерживается вложенность данных любой глубины

Основные операции с данными

Вставка данных

Вставка одного документа:

from tinydb import TinyDB
db = TinyDB('db.json')

# Простая вставка
db.insert({'name': 'Иван', 'age': 30, 'city': 'Москва'})

# Вставка с получением ID
doc_id = db.insert({'name': 'Анна', 'age': 25})
print(f"Документ добавлен с ID: {doc_id}")

Множественная вставка:

users = [
    {'name': 'Анна', 'age': 25, 'profession': 'Дизайнер'},
    {'name': 'Пётр', 'age': 40, 'profession': 'Программист'},
    {'name': 'Мария', 'age': 28, 'profession': 'Менеджер'}
]

doc_ids = db.insert_multiple(users)
print(f"Добавлено документов: {len(doc_ids)}")

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

Получение всех документов:

all_users = db.all()
print(f"Всего пользователей: {len(all_users)}")

Получение документа по ID:

user = db.get(doc_id=1)
if user:
    print(f"Найден пользователь: {user['name']}")

Поиск и фильтрация

Основы поиска с Query:

from tinydb import Query
User = Query()

# Поиск по точному совпадению
users = db.search(User.name == 'Иван')

# Поиск по условию
adults = db.search(User.age >= 30)

# Получение первого совпадения
user = db.get(User.name == 'Анна')

Сложные условия поиска:

# Логическое И
young_designers = db.search((User.age < 30) & (User.profession == 'Дизайнер'))

# Логическое ИЛИ
specialists = db.search((User.profession == 'Программист') | (User.profession == 'Дизайнер'))

# Логическое НЕ
not_managers = db.search(~(User.profession == 'Менеджер'))

Поиск по вложенным полям:

# Добавление документа с вложенной структурой
db.insert({
    'name': 'Алексей',
    'contact': {
        'email': 'alex@example.com',
        'phone': '+7-999-123-45-67'
    },
    'skills': ['Python', 'JavaScript', 'SQL']
})

# Поиск по вложенным полям
user = db.get(User.contact.email == 'alex@example.com')

Обновление данных

Простое обновление:

# Обновление по условию
db.update({'age': 31}, User.name == 'Иван')

# Обновление нескольких полей
db.update({'age': 26, 'city': 'Санкт-Петербург'}, User.name == 'Анна')

Обновление с использованием функций:

# Увеличение возраста на 1
db.update(lambda doc: {'age': doc['age'] + 1}, User.age < 30)

# Добавление нового навыка
def add_skill(doc):
    skills = doc.get('skills', [])
    if 'Docker' not in skills:
        skills.append('Docker')
    return {'skills': skills}

db.update(add_skill, User.name == 'Алексей')

Обновление по doc_id:

db.update({'status': 'active'}, doc_ids=[1, 2, 3])

Удаление данных

Удаление по условию:

# Удаление конкретного пользователя
db.remove(User.name == 'Пётр')

# Удаление по возрасту
db.remove(User.age > 40)

Удаление по doc_id:

db.remove(doc_ids=[1, 2, 3])

Очистка всей таблицы:

db.truncate()

Работа с таблицами

Создание и использование таблиц

TinyDB поддерживает несколько таблиц в одном файле:

from tinydb import TinyDB

db = TinyDB('app.json')

# Создание таблиц
users = db.table('users')
products = db.table('products')
orders = db.table('orders')

# Работа с таблицами
users.insert({'name': 'Иван', 'role': 'admin'})
products.insert({'name': 'Laptop', 'price': 50000})
orders.insert({'user_id': 1, 'product_id': 1, 'quantity': 1})

Получение информации о таблицах

# Список всех таблиц
table_names = db.tables()
print(f"Доступные таблицы: {table_names}")

# Количество документов в таблице
user_count = len(users)
print(f"Пользователей: {user_count}")

# Проверка существования таблицы
if 'users' in db.tables():
    print("Таблица пользователей существует")

Удаление таблиц

# Удаление таблицы
db.drop_table('old_data')

# Удаление всех таблиц
db.drop_tables()

Расширенные возможности поиска

Сравнение и проверка значений

from tinydb import Query

User = Query()

# Числовые сравнения
young_users = db.search(User.age < 25)
middle_aged = db.search(User.age.between(25, 35))

# Строковые операции
name_contains = db.search(User.name.matches(r'.*ан.*'))  # Регулярные выражения
starts_with_a = db.search(User.name.startswith('А'))

# Проверка существования поля
has_email = db.search(User.email.exists())

# Проверка на пустоту
non_empty_skills = db.search(User.skills.exists() & (User.skills != []))

Работа с массивами

# Поиск по элементам массива
python_devs = db.search(User.skills.any(['Python']))
multi_skilled = db.search(User.skills.all(['Python', 'JavaScript']))

# Поиск по длине массива
experienced = db.search(User.skills.test(lambda x: len(x) > 3))

Кастомные функции поиска

# Создание кастомной функции поиска
def is_senior_dev(doc):
    return (doc.get('age', 0) > 30 and 
            'Python' in doc.get('skills', []) and 
            doc.get('experience', 0) > 5)

senior_devs = db.search(User.test(is_senior_dev))

Конфигурация и хранение

Типы хранилищ

from tinydb.storages import JSONStorage, MemoryStorage
from tinydb.middlewares import CachingMiddleware

# Обычное JSON хранилище
db = TinyDB('db.json', storage=JSONStorage)

# Хранилище в памяти
db = TinyDB(storage=MemoryStorage)

# Хранилище с кэшированием
db = TinyDB('db.json', storage=CachingMiddleware(JSONStorage))

Настройка производительности

# Отключение автосохранения для batch операций
db.storage.write_back = False

# Множественные операции
for i in range(1000):
    db.insert({'id': i, 'value': f'item_{i}'})

# Принудительное сохранение
db.storage.flush()
db.storage.write_back = True

Создание кастомного хранилища

from tinydb.storages import Storage
import json
import gzip

class CompressedJSONStorage(Storage):
    def __init__(self, path):
        self.path = path
    
    def read(self):
        try:
            with gzip.open(self.path, 'rt', encoding='utf-8') as f:
                return json.load(f)
        except FileNotFoundError:
            return {}
    
    def write(self, data):
        with gzip.open(self.path, 'wt', encoding='utf-8') as f:
            json.dump(data, f, ensure_ascii=False, indent=2)

# Использование
db = TinyDB('compressed.json.gz', storage=CompressedJSONStorage)

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

Индексирование

# Создание индекса для часто используемых полей
from tinydb.operations import set

# Индекс создаётся автоматически при частом использовании
# Для лучшей производительности используйте числовые ID
db.insert({'id': 1, 'name': 'Иван', 'indexed_field': 'value'})

Пакетные операции

# Эффективная вставка больших объёмов данных
batch_data = [{'id': i, 'value': f'item_{i}'} for i in range(1000)]

# Отключаем автосохранение
with db.storage.write_back_disabled():
    db.insert_multiple(batch_data)

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

# Используйте get() вместо search() для единичных записей
user = db.get(User.id == 123)  # Быстрее
users = db.search(User.id == 123)  # Медленнее для одной записи

# Ограничение результатов
first_10_users = db.search(User.age > 18)[:10]

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

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

from flask import Flask, request, jsonify
from tinydb import TinyDB, Query

app = Flask(__name__)
db = TinyDB('users.json')
User = Query()

@app.route('/users', methods=['GET'])
def get_users():
    users = db.all()
    return jsonify(users)

@app.route('/users', methods=['POST'])
def create_user():
    data = request.get_json()
    doc_id = db.insert(data)
    return jsonify({'id': doc_id, 'message': 'User created'})

@app.route('/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
    user = db.get(doc_id=user_id)
    if user:
        return jsonify(user)
    return jsonify({'error': 'User not found'}), 404

@app.route('/users/<int:user_id>', methods=['PUT'])
def update_user(user_id):
    data = request.get_json()
    db.update(data, doc_ids=[user_id])
    return jsonify({'message': 'User updated'})

@app.route('/users/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
    db.remove(doc_ids=[user_id])
    return jsonify({'message': 'User deleted'})

if __name__ == '__main__':
    app.run(debug=True)

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

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from tinydb import TinyDB, Query
from typing import List, Optional

app = FastAPI()
db = TinyDB('api_users.json')
User = Query()

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

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

@app.post("/users/", response_model=UserResponse)
async def create_user(user: UserCreate):
    doc_id = db.insert(user.dict())
    return UserResponse(id=doc_id, **user.dict())

@app.get("/users/", response_model=List[UserResponse])
async def get_users():
    users = db.all()
    return [UserResponse(id=user.doc_id, **user) for user in users]

@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int):
    user = db.get(doc_id=user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return UserResponse(id=user.doc_id, **user)

Обработка ошибок и безопасность

Обработка ошибок

from tinydb.storages import JSONStorage
from tinydb import TinyDB
import json

try:
    db = TinyDB('db.json')
    user = db.get(User.id == 123)
except json.JSONDecodeError:
    print("Ошибка чтения JSON файла")
    # Создание резервной копии и новой БД
    import shutil
    shutil.move('db.json', 'db.json.backup')
    db = TinyDB('db.json')
except Exception as e:
    print(f"Неожиданная ошибка: {e}")

Валидация данных

from jsonschema import validate, ValidationError

# Схема для валидации пользователей
user_schema = {
    "type": "object",
    "properties": {
        "name": {"type": "string", "minLength": 1},
        "age": {"type": "integer", "minimum": 0},
        "email": {"type": "string", "format": "email"}
    },
    "required": ["name", "age"]
}

def safe_insert(db, document):
    try:
        validate(document, user_schema)
        return db.insert(document)
    except ValidationError as e:
        print(f"Ошибка валидации: {e.message}")
        return None

# Использование
doc_id = safe_insert(db, {'name': 'Иван', 'age': 30, 'email': 'ivan@example.com'})

Блокировка файлов

import fcntl
import json

class LockedJSONStorage:
    def __init__(self, path):
        self.path = path
    
    def read(self):
        try:
            with open(self.path, 'r', encoding='utf-8') as f:
                fcntl.flock(f.fileno(), fcntl.LOCK_SH)
                return json.load(f)
        except FileNotFoundError:
            return {}
    
    def write(self, data):
        with open(self.path, 'w', encoding='utf-8') as f:
            fcntl.flock(f.fileno(), fcntl.LOCK_EX)
            json.dump(data, f, ensure_ascii=False, indent=2)

# Использование с блокировкой
db = TinyDB('locked.json', storage=LockedJSONStorage)

Тестирование с TinyDB

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

import pytest
from tinydb import TinyDB
from tinydb.storages import MemoryStorage

@pytest.fixture
def test_db():
    """Создание тестовой базы данных в памяти"""
    db = TinyDB(storage=MemoryStorage)
    
    # Предварительное заполнение тестовыми данными
    test_users = [
        {'name': 'Тест Юзер 1', 'age': 25, 'role': 'user'},
        {'name': 'Тест Юзер 2', 'age': 30, 'role': 'admin'},
        {'name': 'Тест Юзер 3', 'age': 22, 'role': 'user'}
    ]
    
    db.insert_multiple(test_users)
    yield db
    
    # Очистка после тестов
    db.truncate()

def test_user_creation(test_db):
    """Тест создания пользователя"""
    user_data = {'name': 'Новый Пользователь', 'age': 28, 'role': 'user'}
    doc_id = test_db.insert(user_data)
    
    assert doc_id is not None
    assert len(test_db.all()) == 4  # 3 предустановленных + 1 новый

def test_user_search(test_db):
    """Тест поиска пользователей"""
    from tinydb import Query
    User = Query()
    
    admins = test_db.search(User.role == 'admin')
    assert len(admins) == 1
    assert admins[0]['name'] == 'Тест Юзер 2'

def test_user_update(test_db):
    """Тест обновления пользователя"""
    from tinydb import Query
    User = Query()
    
    test_db.update({'age': 26}, User.name == 'Тест Юзер 1')
    updated_user = test_db.get(User.name == 'Тест Юзер 1')
    
    assert updated_user['age'] == 26

Моки и заглушки

from unittest.mock import patch, MagicMock
import pytest

def test_database_error_handling():
    """Тест обработки ошибок базы данных"""
    with patch('tinydb.TinyDB') as mock_db:
        mock_db.side_effect = Exception("Database connection failed")
        
        with pytest.raises(Exception) as exc_info:
            db = TinyDB('test.json')
        
        assert "Database connection failed" in str(exc_info.value)

Полная справочная таблица методов и функций

Сравнение с другими решениями

Лучшие практики использования

Когда использовать TinyDB

TinyDB идеально подходит для:

• Прототипирования и MVP разработки
• CLI-утилит и скриптов
• Локальных конфигураций приложений
• Тестирования и разработки
• Образовательных проектов
• Простых десктопных приложений
• Хранения небольших объемов данных (до 10,000 записей)

Когда НЕ использовать TinyDB

Избегайте TinyDB при:

• Высоких требованиях к производительности
• Больших объемах данных (более 10,000 записей)
• Необходимости в сложных запросах с JOIN
• Требованиях к ACID транзакциям
• Многопользовательских приложениях с конкурентным доступом
• Критически важных приложениях

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

# Хорошо: Пакетная вставка
users = [{'name': f'User {i}', 'age': 20+i} for i in range(100)]
db.insert_multiple(users)

# Плохо: Множественные одиночные вставки
for i in range(100):
    db.insert({'name': f'User {i}', 'age': 20+i})

# Хорошо: Использование get() для единичных записей
user = db.get(User.id == 123)

# Плохо: Использование search() для единичных записей
users = db.search(User.id == 123)
user = users[0] if users else None

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

# Хорошо: Плоская структура для частых поисков
user = {
    'id': 123,
    'name': 'Иван Петров',
    'email': 'ivan@example.com',
    'age': 30,
    'city': 'Москва',
    'role': 'admin'
}

# Приемлемо: Вложенная структура для связанных данных
user = {
    'id': 123,
    'name': 'Иван Петров',
    'contact': {
        'email': 'ivan@example.com',
        'phone': '+7-999-123-45-67'
    },
    'profile': {
        'age': 30,
        'city': 'Москва',
        'bio': 'Разработчик Python'
    }
}

Миграции и версионирование

Система версионирования базы данных

from tinydb import TinyDB, Query

class DatabaseMigrator:
    def __init__(self, db_path):
        self.db = TinyDB(db_path)
        self.meta = self.db.table('_meta')
        
    def get_version(self):
        Meta = Query()
        version_doc = self.meta.get(Meta.key == 'version')
        return version_doc['value'] if version_doc else 0
    
    def set_version(self, version):
        Meta = Query()
        self.meta.upsert({'key': 'version', 'value': version}, Meta.key == 'version')
    
    def migrate_to_v2(self):
        """Добавление поля 'status' всем пользователям"""
        users = self.db.table('users')
        User = Query()
        
        for user in users.all():
            if 'status' not in user:
                users.update({'status': 'active'}, doc_ids=[user.doc_id])
    
    def migrate_to_v3(self):
        """Преобразование поля 'age' в 'birth_year'"""
        users = self.db.table('users')
        from datetime import datetime
        
        current_year = datetime.now().year
        
        for user in users.all():
            if 'age' in user and 'birth_year' not in user:
                birth_year = current_year - user['age']
                users.update({'birth_year': birth_year}, doc_ids=[user.doc_id])
                # Удаляем старое поле
                new_user = {k: v for k, v in user.items() if k != 'age'}
                users.update(new_user, doc_ids=[user.doc_id])
    
    def migrate(self):
        current_version = self.get_version()
        
        if current_version < 2:
            print("Миграция до версии 2...")
            self.migrate_to_v2()
            self.set_version(2)
        
        if current_version < 3:
            print("Миграция до версии 3...")
            self.migrate_to_v3()
            self.set_version(3)
        
        print(f"База данных обновлена до версии {self.get_version()}")

# Использование
migrator = DatabaseMigrator('app.json')
migrator.migrate()

Резервное копирование и восстановление

Создание резервных копий

import shutil
import json
from datetime import datetime
from tinydb import TinyDB

class BackupManager:
    def __init__(self, db_path):
        self.db_path = db_path
        self.db = TinyDB(db_path)
    
    def create_backup(self, backup_dir='backups'):
        """Создание резервной копии"""
        import os
        os.makedirs(backup_dir, exist_ok=True)
        
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
        backup_name = f"backup_{timestamp}.json"
        backup_path = os.path.join(backup_dir, backup_name)
        
        shutil.copy2(self.db_path, backup_path)
        return backup_path
    
    def restore_backup(self, backup_path):
        """Восстановление из резервной копии"""
        self.db.close()
        shutil.copy2(backup_path, self.db_path)
        self.db = TinyDB(self.db_path)
    
    def export_to_json(self, export_path):
        """Экспорт данных в JSON"""
        data = {
            'tables': {},
            'timestamp': datetime.now().isoformat()
        }
        
        for table_name in self.db.tables():
            table = self.db.table(table_name)
            data['tables'][table_name] = table.all()
        
        with open(export_path, 'w', encoding='utf-8') as f:
            json.dump(data, f, ensure_ascii=False, indent=2)
    
    def import_from_json(self, import_path):
        """Импорт данных из JSON"""
        with open(import_path, 'r', encoding='utf-8') as f:
            data = json.load(f)
        
        for table_name, documents in data['tables'].items():
            table = self.db.table(table_name)
            table.truncate()
            table.insert_multiple(documents)

# Использование
backup_manager = BackupManager('app.json')
backup_path = backup_manager.create_backup()
print(f"Резервная копия создана: {backup_path}")

Частые вопросы и решения

Почему TinyDB медленно работает с большими данными?

TinyDB загружает весь файл JSON в память при каждой операции. Для больших объемов данных рекомендуется использовать SQLite или PostgreSQL.

Как обеспечить многопоточную безопасность?

import threading
from tinydb import TinyDB

class ThreadSafeTinyDB:
    def __init__(self, *args, **kwargs):
        self.db = TinyDB(*args, **kwargs)
        self.lock = threading.Lock()
    
    def insert(self, *args, **kwargs):
        with self.lock:
            return self.db.insert(*args, **kwargs)
    
    def search(self, *args, **kwargs):
        with self.lock:
            return self.db.search(*args, **kwargs)
    
    def update(self, *args, **kwargs):
        with self.lock:
            return self.db.update(*args, **kwargs)
    
    def remove(self, *args, **kwargs):
        with self.lock:
            return self.db.remove(*args, **kwargs)

# Использование
db = ThreadSafeTinyDB('thread_safe.json')

Как валидировать данные перед вставкой?

from jsonschema import validate, ValidationError

def validate_user(user_data):
    schema = {
        "type": "object",
        "properties": {
            "name": {"type": "string", "minLength": 1},
            "age": {"type": "integer", "minimum": 0, "maximum": 150},
            "email": {"type": "string", "format": "email"}
        },
        "required": ["name", "age"]
    }
    
    try:
        validate(user_data, schema)
        return True
    except ValidationError as e:
        print(f"Ошибка валидации: {e.message}")
        return False

# Использование
user_data = {'name': 'Иван', 'age': 30, 'email': 'ivan@example.com'}
if validate_user(user_data):
    db.insert(user_data)

Как сделать автоматическое резервное копирование?

import schedule
import time
from datetime import datetime

def auto_backup():
    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
    backup_path = f'backups/auto_backup_{timestamp}.json'
    
    try:
        backup_manager.create_backup()
        print(f"Автоматическая резервная копия создана: {backup_path}")
    except Exception as e:
        print(f"Ошибка создания резервной копии: {e}")

# Настройка автоматического резервного копирования каждые 6 часов
schedule.every(6).hours.do(auto_backup)

# Запуск в отдельном потоке
import threading

def backup_scheduler():
    while True:
        schedule.run_pending()
        time.sleep(60)

backup_thread = threading.Thread(target=backup_scheduler, daemon=True)
backup_thread.start()

Как оптимизировать производительность запросов?

# Кэширование часто используемых запросов
from functools import lru_cache

@lru_cache(maxsize=128)
def get_user_by_email(email):
    return db.get(User.email == email)

# Использование индексов для поиска
# Создайте отдельное поле для индексирования
db.insert({
    'name': 'Иван',
    'email': 'ivan@example.com',
    'email_hash': hash('ivan@example.com')  # Быстрый поиск по хешу
})

# Ограничение результатов
recent_users = db.search(User.created_at > yesterday)[:100]

Как обрабатывать ошибки повреждения JSON?

import json
import shutil
from tinydb import TinyDB

def safe_load_db(db_path):
    try:
        db = TinyDB(db_path)
        # Проверяем, что файл читается корректно
        db.all()
        return db
    except json.JSONDecodeError:
        print("Файл базы данных поврежден!")
        
        # Создаем резервную копию поврежденного файла
        backup_path = f"{db_path}.corrupted.backup"
        shutil.copy2(db_path, backup_path)
        print(f"Поврежденный файл сохранен как: {backup_path}")
        
        # Создаем новую базу данных
        db = TinyDB(db_path)
        return db

# Использование
db = safe_load_db('app.json')

Заключение

TinyDB представляет собой отличное решение для проектов, где важны простота использования и минимальные зависимости. Эта легковесная NoSQL база данных идеально подходит для прототипирования, CLI-утилит, тестирования и небольших приложений.

Главные преимущества TinyDB:

• Простота установки и использования
• Нулевая конфигурация
• Читаемый формат хранения данных (JSON)
• Отсутствие зависимостей от внешних сервисов
• Интуитивно понятный API
• Полная интеграция с экосистемой Python

При правильном использовании TinyDB может значительно ускорить разработку, особенно на этапе прототипирования и создания MVP. Однако важно помнить о её ограничениях и не использовать для высоконагруженных приложений или больших объемов данных.

Благодаря своей простоте и эффективности, TinyDB остается популярным выбором для разработчиков, которым нужно быстро создать рабочее решение без лишней сложности.