Custom Storage Framework

class ICustomStorage:
    """Interface for custom storage implementations."""
    
    # Error constants
    NoError = 0
    InvalidPageError = 1
    IllegalStateError = 2
    
    # Page constants  
    EmptyPage = -0x1
    NewPage = -0x1
    
    def registerCallbacks(self, properties):
        """
        Register storage callbacks with index properties.
        
        Parameters:
        - properties (Property): Property object to configure
        
        Abstract method that must be implemented by subclasses.
        """
    
    def clear(self):
        """
        Clear all stored data.
        
        Abstract method that must be implemented by subclasses.
        """
    
    @property
    def hasData(self) -> bool:
        """
        Indicate whether storage contains data.
        
        Returns:
        bool: True if storage has existing data
        """
    
    def allocateBuffer(self, length: int):
        """
        Allocate buffer for data operations.
        
        Parameters:
        - length (int): Buffer size in bytes
        
        Returns:
        Buffer object for data operations
        """

class CustomStorageBase(ICustomStorage):
    """Base class for raw C buffer access custom storage."""
    
    def create(self, context, returnError):
        """
        Create storage backend.
        
        Parameters:
        - context: Storage context object
        - returnError: Error reporting callback
        
        Abstract method for storage initialization.
        """
    
    def destroy(self, context, returnError):
        """
        Destroy storage backend and cleanup resources.
        
        Parameters:
        - context: Storage context object  
        - returnError: Error reporting callback
        
        Abstract method for storage cleanup.
        """
    
    def loadByteArray(self, context, page, resultLen, resultData, returnError):
        """
        Load data from storage backend.
        
        Parameters:
        - context: Storage context object
        - page (int): Page identifier to load
        - resultLen: Output parameter for data length
        - resultData: Output parameter for data buffer
        - returnError: Error reporting callback
        
        Abstract method for data loading.
        """
    
    def storeByteArray(self, context, page, len, data, returnError):
        """
        Store data to storage backend.
        
        Parameters:
        - context: Storage context object
        - page (int): Page identifier for storage
        - len (int): Data length in bytes
        - data: Data buffer to store
        - returnError: Error reporting callback
        
        Abstract method for data storage.
        """
    
    def deleteByteArray(self, context, page, returnError):
        """
        Delete data from storage backend.
        
        Parameters:
        - context: Storage context object
        - page (int): Page identifier to delete
        - returnError: Error reporting callback
        
        Abstract method for data deletion.
        """
    
    def flush(self, context, returnError):
        """
        Flush pending operations to storage backend.
        
        Parameters:
        - context: Storage context object
        - returnError: Error reporting callback
        
        Abstract method for storage flushing.
        """

class CustomStorage(CustomStorageBase):
    """Default custom storage with Python string marshaling."""
    
    def create(self, returnError):
        """
        Create storage backend.
        
        Parameters:
        - returnError: Error reporting callback
        
        Abstract method for storage initialization.
        Simplified interface without context parameter.
        """
    
    def destroy(self, returnError):
        """
        Destroy storage backend and cleanup resources.
        
        Parameters:
        - returnError: Error reporting callback
        
        Abstract method for storage cleanup.
        Simplified interface without context parameter.
        """
    
    def loadByteArray(self, page, returnError):
        """
        Load data from storage backend as Python string.
        
        Parameters:
        - page (int): Page identifier to load
        - returnError: Error reporting callback
        
        Returns:
        str: Loaded data as Python string
        
        Abstract method for data loading with automatic marshaling.
        """
    
    def storeByteArray(self, page, data, returnError):
        """
        Store Python string data to storage backend.
        
        Parameters:
        - page (int): Page identifier for storage
        - data (str): Python string data to store
        - returnError: Error reporting callback
        
        Abstract method for data storage with automatic marshaling.
        """
    
    def deleteByteArray(self, page, returnError):
        """
        Delete data from storage backend.
        
        Parameters:
        - page (int): Page identifier to delete
        - returnError: Error reporting callback
        
        Abstract method for data deletion.
        """
    
    def flush(self, returnError):
        """
        Flush pending operations to storage backend.
        
        Parameters:
        - returnError: Error reporting callback
        
        Abstract method for storage flushing.
        """

from rtree.index import Index, Property, CustomStorage

class DictStorage(CustomStorage):
    """Simple dictionary-based storage for demonstration."""
    
    def __init__(self):
        super().__init__()
        self.data = {}
        self.has_data = False
    
    def create(self, returnError):
        """Initialize storage."""
        self.data.clear()
        self.has_data = True
        returnError.value = self.NoError
    
    def destroy(self, returnError):
        """Cleanup storage."""
        self.data.clear()
        self.has_data = False
        returnError.value = self.NoError
    
    def loadByteArray(self, page, returnError):
        """Load data for page."""
        if page in self.data:
            returnError.value = self.NoError
            return self.data[page]
        else:
            returnError.value = self.InvalidPageError
            return None
    
    def storeByteArray(self, page, data, returnError):
        """Store data for page."""
        self.data[page] = data
        returnError.value = self.NoError
    
    def deleteByteArray(self, page, returnError):
        """Delete data for page."""
        if page in self.data:
            del self.data[page]
            returnError.value = self.NoError
        else:
            returnError.value = self.InvalidPageError
    
    def flush(self, returnError):
        """Flush operations (no-op for dict)."""
        returnError.value = self.NoError
    
    @property
    def hasData(self):
        return self.has_data

# Use custom storage
storage = DictStorage()
prop = Property(storage=RT_Custom)
storage.registerCallbacks(prop)

idx = Index(properties=prop, storage=storage)
idx.insert(0, (0, 0, 10, 10))

import sqlite3
from rtree.index import Index, Property, CustomStorage

class SQLiteStorage(CustomStorage):
    """SQLite database storage backend."""
    
    def __init__(self, db_path=":memory:"):
        super().__init__()
        self.db_path = db_path
        self.conn = None
    
    def create(self, returnError):
        """Initialize database storage."""
        try:
            self.conn = sqlite3.connect(self.db_path)
            self.conn.execute("""
                CREATE TABLE IF NOT EXISTS rtree_pages (
                    page_id INTEGER PRIMARY KEY,
                    data BLOB
                )
            """)
            self.conn.commit()
            returnError.value = self.NoError
        except Exception:
            returnError.value = self.IllegalStateError
    
    def destroy(self, returnError):
        """Cleanup database connection."""
        try:
            if self.conn:
                self.conn.close()
                self.conn = None
            returnError.value = self.NoError
        except Exception:
            returnError.value = self.IllegalStateError
    
    def loadByteArray(self, page, returnError):
        """Load page data from database."""
        try:
            cursor = self.conn.execute(
                "SELECT data FROM rtree_pages WHERE page_id = ?", (page,)
            )
            row = cursor.fetchone()
            if row:
                returnError.value = self.NoError
                return row[0].decode('utf-8')
            else:
                returnError.value = self.InvalidPageError
                return None
        except Exception:
            returnError.value = self.IllegalStateError
            return None
    
    def storeByteArray(self, page, data, returnError):
        """Store page data to database."""
        try:
            self.conn.execute(
                "INSERT OR REPLACE INTO rtree_pages (page_id, data) VALUES (?, ?)",
                (page, data.encode('utf-8'))
            )
            returnError.value = self.NoError
        except Exception:
            returnError.value = self.IllegalStateError
    
    def deleteByteArray(self, page, returnError):
        """Delete page data from database."""
        try:
            cursor = self.conn.execute(
                "DELETE FROM rtree_pages WHERE page_id = ?", (page,)
            )
            if cursor.rowcount > 0:
                returnError.value = self.NoError
            else:
                returnError.value = self.InvalidPageError
        except Exception:
            returnError.value = self.IllegalStateError
    
    def flush(self, returnError):
        """Commit pending transactions."""
        try:
            self.conn.commit()
            returnError.value = self.NoError
        except Exception:
            returnError.value = self.IllegalStateError
    
    @property
    def hasData(self):
        if not self.conn:
            return False
        cursor = self.conn.execute("SELECT COUNT(*) FROM rtree_pages")
        return cursor.fetchone()[0] > 0

# Use SQLite storage
storage = SQLiteStorage("spatial_index.db")
prop = Property(storage=RT_Custom)
storage.registerCallbacks(prop)

idx = Index(properties=prop, storage=storage)

import requests
from rtree.index import Index, Property, CustomStorage

class HTTPStorage(CustomStorage):
    """HTTP-based storage backend for distributed indexing."""
    
    def __init__(self, base_url, auth_token=None):
        super().__init__()
        self.base_url = base_url.rstrip('/')
        self.auth_token = auth_token
        self.headers = {}
        if auth_token:
            self.headers['Authorization'] = f'Bearer {auth_token}'
    
    def create(self, returnError):
        """Initialize remote storage."""
        try:
            response = requests.post(
                f"{self.base_url}/storage/create",
                headers=self.headers
            )
            if response.status_code == 200:
                returnError.value = self.NoError
            else:
                returnError.value = self.IllegalStateError
        except Exception:
            returnError.value = self.IllegalStateError
    
    def destroy(self, returnError):
        """Cleanup remote storage."""
        try:
            response = requests.delete(
                f"{self.base_url}/storage",
                headers=self.headers
            )
            returnError.value = self.NoError
        except Exception:
            returnError.value = self.IllegalStateError
    
    def loadByteArray(self, page, returnError):
        """Load page from remote storage."""
        try:
            response = requests.get(
                f"{self.base_url}/storage/pages/{page}",
                headers=self.headers
            )
            if response.status_code == 200:
                returnError.value = self.NoError
                return response.text
            elif response.status_code == 404:
                returnError.value = self.InvalidPageError
                return None
            else:
                returnError.value = self.IllegalStateError
                return None
        except Exception:
            returnError.value = self.IllegalStateError
            return None
    
    def storeByteArray(self, page, data, returnError):
        """Store page to remote storage."""
        try:
            response = requests.put(
                f"{self.base_url}/storage/pages/{page}",
                data=data,
                headers=self.headers
            )
            if response.status_code in (200, 201):
                returnError.value = self.NoError
            else:
                returnError.value = self.IllegalStateError
        except Exception:
            returnError.value = self.IllegalStateError
    
    def deleteByteArray(self, page, returnError):
        """Delete page from remote storage."""
        try:
            response = requests.delete(
                f"{self.base_url}/storage/pages/{page}",
                headers=self.headers
            )
            if response.status_code in (200, 204):
                returnError.value = self.NoError
            elif response.status_code == 404:
                returnError.value = self.InvalidPageError
            else:
                returnError.value = self.IllegalStateError
        except Exception:
            returnError.value = self.IllegalStateError
    
    def flush(self, returnError):
        """Flush pending operations."""
        try:
            response = requests.post(
                f"{self.base_url}/storage/flush",
                headers=self.headers
            )
            returnError.value = self.NoError
        except Exception:
            returnError.value = self.IllegalStateError
    
    @property
    def hasData(self):
        try:
            response = requests.head(
                f"{self.base_url}/storage",
                headers=self.headers
            )
            return response.status_code == 200
        except Exception:
            return False

# Use HTTP storage
storage = HTTPStorage("https://api.spatial-service.com", "your-auth-token")
prop = Property(storage=RT_Custom)
storage.registerCallbacks(prop)

idx = Index(properties=prop, storage=storage)

import tempfile
import os
from rtree.index import Index, Property, CustomStorage

class CachedFileStorage(CustomStorage):
    """File storage with in-memory caching."""
    
    def __init__(self, base_dir=None, cache_size=1000):
        super().__init__()
        self.base_dir = base_dir or tempfile.mkdtemp()
        self.cache = {}
        self.cache_size = cache_size
        self.access_order = []
    
    def _cache_key(self, page):
        return f"page_{page}"
    
    def _evict_cache(self):
        """Evict oldest entries if cache is full."""
        while len(self.cache) >= self.cache_size:
            oldest = self.access_order.pop(0)
            if oldest in self.cache:
                del self.cache[oldest]
    
    def _update_access(self, key):
        """Update access order for LRU eviction."""
        if key in self.access_order:
            self.access_order.remove(key)
        self.access_order.append(key)
    
    def create(self, returnError):
        """Initialize file storage."""
        try:
            os.makedirs(self.base_dir, exist_ok=True)
            returnError.value = self.NoError
        except Exception:
            returnError.value = self.IllegalStateError
    
    def destroy(self, returnError):
        """Cleanup file storage."""
        try:
            self.cache.clear()
            self.access_order.clear()
            returnError.value = self.NoError
        except Exception:
            returnError.value = self.IllegalStateError
    
    def loadByteArray(self, page, returnError):
        """Load page with caching."""
        cache_key = self._cache_key(page)
        
        # Check cache first
        if cache_key in self.cache:
            self._update_access(cache_key)
            returnError.value = self.NoError
            return self.cache[cache_key]
        
        # Load from file
        try:
            file_path = os.path.join(self.base_dir, f"page_{page}.dat")
            if os.path.exists(file_path):
                with open(file_path, 'r') as f:
                    data = f.read()
                
                # Cache the data
                self._evict_cache()
                self.cache[cache_key] = data
                self._update_access(cache_key)
                
                returnError.value = self.NoError
                return data
            else:
                returnError.value = self.InvalidPageError
                return None
        except Exception:
            returnError.value = self.IllegalStateError
            return None
    
    def storeByteArray(self, page, data, returnError):
        """Store page with caching."""
        try:
            # Write to file
            file_path = os.path.join(self.base_dir, f"page_{page}.dat")
            with open(file_path, 'w') as f:
                f.write(data)
            
            # Update cache
            cache_key = self._cache_key(page)
            self._evict_cache()
            self.cache[cache_key] = data
            self._update_access(cache_key)
            
            returnError.value = self.NoError
        except Exception:
            returnError.value = self.IllegalStateError
    
    def deleteByteArray(self, page, returnError):
        """Delete page and remove from cache."""
        try:
            file_path = os.path.join(self.base_dir, f"page_{page}.dat")
            if os.path.exists(file_path):
                os.remove(file_path)
                
                # Remove from cache
                cache_key = self._cache_key(page)
                if cache_key in self.cache:
                    del self.cache[cache_key]
                    self.access_order.remove(cache_key)
                
                returnError.value = self.NoError
            else:
                returnError.value = self.InvalidPageError
        except Exception:
            returnError.value = self.IllegalStateError
    
    def flush(self, returnError):
        """Flush operation (files are written immediately)."""
        returnError.value = self.NoError
    
    @property
    def hasData(self):
        try:
            return len(os.listdir(self.base_dir)) > 0
        except Exception:
            return False

# Use cached file storage
storage = CachedFileStorage("/tmp/rtree_cache", cache_size=500)
prop = Property(storage=RT_Custom)
storage.registerCallbacks(prop)

idx = Index(properties=prop, storage=storage)

class CustomStorageCallbacks:
    """C callback functions for custom storage (ctypes.Structure)."""
    # Internal structure for C API integration
    # Used by registerCallbacks() to configure storage backend

# Error codes for custom storage
NoError = 0
InvalidPageError = 1
IllegalStateError = 2

# Special page identifiers
EmptyPage = -0x1
NewPage = -0x1