Skip to main content

​
Notion Repository

A clean, efficient, and maintainable Python library for interacting with Notion databases. Built with async/await patterns, intelligent caching, and modular design.

​
🚀 Features

  • Fully Async: Built from the ground up with async/await for optimal performance
  • Smart Caching: Intelligent lookup table management with Django cache integration
  • Type Safe: Complete type annotations for better IDE support and code reliability
  • Modular Design: Easily extensible architecture for new data types
  • Error Resilient: Comprehensive error handling and logging
  • Performance Optimized: Automatic pagination, batch operations, and on-demand loading

​
📋 Table of Contents

​
🛠️ Installation

pip install -r requirements.txt
Requirements:
  • Python 3.8+
  • Django (for caching)
  • aiohttp or requests
  • Your existing core.utils.ApiClient

​
⚡ Quick Start

​
Basic Setup

from vault.api.trace.notion.repository import new_notion_repository

# Initialize the repository
repo = new_notion_repository(
    notion_token="your_notion_integration_token",
    feature_db_id="your_features_database_id",
    gateway_db_id="your_gateways_database_id",
    tier_db_id="your_tiers_database_id",
    tag_db_id="your_tags_database_id",
    keyword_db_id="your_keywords_database_id"
)

# Fetch data
features = await repo.fetch_features()
gateways = await repo.fetch_gateways(tier_filter="premium")
tiers = await repo.fetch_tiers()

​
With Filters

# Fetch features with filters
api_features = await repo.fetch_features(
    category_filter="API",
    status_filter="Active"
)

# Fetch gateways for specific tier
premium_gateways = await repo.fetch_gateways(
    tier_filter="premium",
    status_filter="live"
)

# Fetch tags by category
seo_tags = await repo.fetch_tags(
    category_filter="SEO",
    type_filter="keyword"
)

​
⚙️ Configuration

​
Database Configuration

The repository requires Notion database IDs for each data type: 
DATABASE_CONFIG = {
    'gateway_db_id': 'your_gateways_database_id',
    'feature_db_id': 'your_features_database_id',
    'tier_db_id': 'your_tiers_database_id',
    'tag_db_id': 'your_tags_database_id',
    'keyword_db_id': 'your_keywords_database_id',
    'grade_db_id': 'your_grades_database_id',
    'status_db_id': 'your_status_database_id'
}

repo = new_notion_repository(
    notion_token="your_token",
    **DATABASE_CONFIG
)

​
Environment Variables

For production, use environment variables: 
import os

repo = new_notion_repository(
    notion_token=os.getenv('NOTION_TOKEN'),
    feature_db_id=os.getenv('NOTION_FEATURES_DB_ID'),
    gateway_db_id=os.getenv('NOTION_IO_DB_ID'),
    # ... other database IDs
)

​
Cache Configuration

Customize cache behavior in your Django settings: 
# settings.py
CACHES = {
    'default': {
        'BACKEND': 'django.core.cache.backends.redis.RedisCache',
        'LOCATION': 'redis://127.0.0.1:6379/1',
        'TIMEOUT': 3600,  # Default cache timeout
    }
}

​
🧠 Core Concepts

​
Architecture Overview

NotionRepository
├── NotionApiClient (handles HTTP communication)
├── LookupManager (manages reference data caching)
└── Fetchers (convert Notion data to Python objects)
    ├── FeatureFetcher
    ├── GatewayFetcher
    ├── TierFetcher
    ├── TagFetcher
    └── KeywordFetcher

​
Lookup Tables

The system automatically manages “lookup tables” - cached mappings of Notion page IDs to human-readable values. This dramatically improves performance by avoiding redundant API calls for reference data. Example: Instead of making API calls for every tier reference, the system caches: 
{
    "page_id_123": "premium",
    "page_id_456": "basic",
    "page_id_789": "enterprise"
}

​
Data Models

Each fetcher returns strongly-typed Python objects:
  • Feature - Product features with metadata
  • Gateway - API gateways and specifications
  • Tier - Subscription tiers and limits
  • Tag - Classification tags with SEO data
  • Keyword - SEO keywords with analytics

​
📚 API Reference

​
NotionRepository

​
Core Methods

async def fetch_features(
    category_filter: Optional[str] = None,
    status_filter: Optional[str] = None,
    owner_filter: Optional[str] = None
) -> List[Feature]

async def fetch_gateways(
    tier_filter: Optional[str] = None,
    feature_filter: Optional[str] = None,
    status_filter: Optional[str] = None
) -> List[Gateway]

async def fetch_tiers(
    tier_filter: Optional[str] = None
) -> List[Tier]

async def fetch_tags(
    tag_filter: Optional[str] = None,
    type_filter: Optional[str] = None,
    category_filter: Optional[str] = None
) -> List[Tag]

async def fetch_keywords(
    keyword_filter: Optional[str] = None
) -> List[Keyword]

​
Utility Methods

async def preload_lookups(
    lookup_names: Optional[List[str]] = None
) -> None
Pre-load lookup tables for better performance. 
async def clear_lookup_cache(
    lookup_name: Optional[str] = None
) -> None
Clear cached lookup data. 
async def get_lookup_stats() -> Dict[str, int]
Get statistics about loaded lookups. 
async def health_check() -> Dict[str, Any]
Comprehensive health check of the repository.

​
🚀 Advanced Usage

​
Performance Optimization

# Pre-load lookups for better performance
await repo.preload_lookups(['tier', 'status', 'tag'])

# Now subsequent calls will be faster
features = await repo.fetch_features()
gateways = await repo.fetch_gateways()

​
Batch Operations

import asyncio

# Fetch multiple data types concurrently
features_task = repo.fetch_features()
gateways_task = repo.fetch_gateways()
tiers_task = repo.fetch_tiers()

features, gateways, tiers = await asyncio.gather(
    features_task,
    gateways_task,
    tiers_task
)

​
Custom Queries

# Raw database queries for advanced use cases
raw_data = await repo.query_database_raw(
    db_id="your_database_id",
    payload={
        "filter": {
            "property": "Status",
            "select": {"equals": "Active"}
        },
        "sorts": [
            {
                "property": "Created",
                "direction": "descending"
            }
        ]
    }
)

​
Error Handling

try:
    features = await repo.fetch_features()
except Exception as e:
    logger.error(f"Failed to fetch features: {e}")
    # Handle error appropriately
    features = []

​
Context Manager Usage

# The repository uses context managers internally for safety
async with repo._ensure_initialized():
    # All operations are safely wrapped
    features = await repo.fetch_features()

​
💡 Performance Tips

​
1. Preload Lookups

# Do this once at application startup
await repo.preload_lookups()

​
2. Use Filters Wisely

# More efficient - filters on Notion side
features = await repo.fetch_features(category_filter="API")

# Less efficient - filters after fetching all data
all_features = await repo.fetch_features()
api_features = [f for f in all_features if f.category == "API"]

# Good - batch related operations
features = await repo.fetch_features()
for feature in features:
    # Process features
    pass

# Avoid - multiple small queries in loops
for feature_id in feature_ids:
    feature = await repo.fetch_features(feature_filter=feature_id)

​
4. Monitor Cache Usage

# Check cache effectiveness
stats = await repo.get_lookup_stats()
print(f"Loaded lookups: {stats}")

# Health check includes cache status
health = await repo.health_check()
print(f"Cache status: {health['lookup_cache_status']}")

​
🔧 Troubleshooting

​
Common Issues

​
1. Missing Database IDs

Error: "Database ID not configured for FeatureFetcher"
Solution: Ensure all required database IDs are provided during initialization.

​
2. Cache Issues

# Clear problematic cache
await repo.clear_lookup_cache('tier')

# Or clear all caches
await repo.clear_lookup_cache()

​
3. API Rate Limits

The repository handles pagination automatically, but for rate limits: 
# Add delays between operations if needed
import asyncio

features = await repo.fetch_features()
await asyncio.sleep(1)  # Rate limit protection
gateways = await repo.fetch_gateways()

​
4. Connection Issues

# Check repository health
health = await repo.health_check()
if health['status'] != 'healthy':
    print(f"Issues detected: {health['errors']}")

​
Debug Mode

Enable debug logging: 
import logging
logging.getLogger('vault.api.trace.notion').setLevel(logging.DEBUG)

​
Health Monitoring

# Regular health checks in production
async def monitor_notion_health():
    health = await repo.health_check()

    if health['status'] != 'healthy':
        # Alert or log issues
        logger.warning(f"Notion repository issues: {health['errors']}")

    return health

​
🧪 Testing

​
Unit Tests

import pytest
from unittest.mock import AsyncMock, MagicMock

@pytest.mark.asyncio
async def test_fetch_features():
    # Mock the API client
    mock_api = AsyncMock()
    mock_api.query_database.return_value = {
        'results': [
            {
                'id': 'page_123',
                'properties': {
                    'feature': {'title': [{'plain_text': 'Test Feature'}]}
                }
            }
        ]
    }

    # Test the fetcher
    repo = new_notion_repository(notion_token="test_token")
    repo.api_client = mock_api

    features = await repo.fetch_features()
    assert len(features) == 1
    assert features[0].feature == "Test Feature"

​
Integration Tests

@pytest.mark.asyncio
async def test_notion_integration():
    """Test with real Notion API (requires test database)"""
    repo = new_notion_repository(
        notion_token=os.getenv('NOTION_TEST_TOKEN'),
        feature_db_id=os.getenv('NOTION_TEST_FEATURES_DB')
    )

    features = await repo.fetch_features()
    assert isinstance(features, list)

​
🤝 Contributing

​
Adding New Data Types

  1. Create a new fetcher class:
class MyDataFetcher(BaseNotionFetcher):
    async def fetch(self, db_id: str, **filters) -> List[MyData]:
        # Implementation
        pass

    def _page_to_mydata(self, page: Dict) -> Optional[MyData]:
        # Convert Notion page to your data model
        pass
  1. Add to repository:
# In NotionRepository.__init__
self.mydata_fetcher = MyDataFetcher(self.api_client, self.lookup_manager, self.fmt)

# Add public method
async def fetch_mydata(self, **filters) -> List[MyData]:
    return await self.mydata_fetcher.fetch(self.db_config.mydata_db_id, **filters)
  1. Update configuration:
# Add to DatabaseConfig
mydata_db_id: Optional[str] = None

​
Code Style

  • Follow PEP 8
  • Use type hints
  • Add docstrings for public methods
  • Include logging for important operations
  • Handle errors gracefully

​
Pull Request Process

  1. Fork the repository
  2. Create a feature branch
  3. Add tests for new functionality
  4. Ensure all tests pass
  5. Update documentation
  6. Submit pull request

​
📄 License

This project is licensed under the MIT License - see the LICENSE file for details.

​
🆘 Support

For issues and questions:
  1. Check the troubleshooting section
  2. Review the API reference
  3. Open an issue on GitHub
  4. Contact the development team
Happy coding! 🎉