Storage and Resumability

from abc import ABC, abstractmethod

class Storage(ABC):
    """Abstract base class for URL storage implementations."""
    
    @abstractmethod
    def get_item(self, key: str) -> Optional[str]:
        """
        Return the tus url of a file, identified by the key specified.
        
        Parameters:
        - key (str): The unique id for the stored item (upload URL)
        
        Returns:
        Optional[str]: The stored URL or None if not found
        """
    
    @abstractmethod
    def set_item(self, key: str, value: str):
        """
        Store the url value under the unique key.
        
        Parameters:
        - key (str): The unique id to store the URL under
        - value (str): The actual URL value to be stored
        """
    
    @abstractmethod
    def remove_item(self, key: str):
        """
        Remove/Delete the url value under the unique key from storage.
        
        Parameters:
        - key (str): The unique id of the item to remove
        """

class FileStorage(Storage):
    """File-based implementation of Storage interface using TinyDB."""
    
    def __init__(self, fp: str):
        """
        Initialize FileStorage with database file path.
        
        Parameters:
        - fp (str): File path for the TinyDB database
        """
    
    def get_item(self, key: str) -> Optional[str]:
        """
        Return the tus url of a file, identified by the key specified.
        
        Parameters:
        - key (str): The unique id for the stored item
        
        Returns:
        Optional[str]: The stored URL or None if not found
        """
    
    def set_item(self, key: str, url: str):
        """
        Store the url value under the unique key.
        
        Updates existing entry if key already exists, otherwise creates new entry.
        
        Parameters:
        - key (str): The unique id to store the URL under  
        - url (str): The actual URL value to be stored
        """
    
    def remove_item(self, key: str):
        """
        Remove/Delete the url value under the unique key from storage.
        
        Parameters:
        - key (str): The unique id of the item to remove
        """
    
    def close(self):
        """
        Close the file storage and release all opened files.
        
        Should be called when done with storage to ensure proper cleanup.
        """

from abc import ABC, abstractmethod
from typing import IO

class Fingerprint(ABC):
    """An interface specifying the requirements of a file fingerprint."""
    
    @abstractmethod
    def get_fingerprint(self, fs: IO) -> str:
        """
        Return a unique fingerprint string value based on the file stream received.
        
        Parameters:
        - fs (IO): The file stream instance to generate fingerprint for
        
        Returns:
        str: Unique fingerprint string identifying the file
        """

class Fingerprint(Fingerprint):
    """MD5-based fingerprint implementation using file content and size."""
    
    BLOCK_SIZE = 65536  # Block size for reading file content
    
    def get_fingerprint(self, fs: IO) -> str:
        """
        Return a unique fingerprint string based on MD5 hash and file size.
        
        Generates fingerprint using MD5 hash of first block and file size
        to minimize collision chances while being efficient.
        
        Parameters:
        - fs (IO): The file stream instance to generate fingerprint for
        
        Returns:
        str: Fingerprint in format "size:{size}--md5:{hash}"
        """

from tusclient import client
from tusclient.storage.filestorage import FileStorage

# Create file storage for URL persistence
storage = FileStorage('/tmp/tuspy_uploads.db')

# Create client and uploader with resumability enabled
my_client = client.TusClient('http://tusd.tusdemo.net/files/')
uploader = my_client.uploader(
    '/path/to/large_file.ext',
    chunk_size=5*1024*1024,  # 5MB chunks
    store_url=True,          # Enable URL storage
    url_storage=storage      # Provide storage implementation
)

try:
    # Upload will be resumable if interrupted
    uploader.upload()
    print("Upload completed successfully")
except KeyboardInterrupt:
    print("Upload interrupted - can be resumed later")
finally:
    storage.close()

from tusclient import client
from tusclient.storage.filestorage import FileStorage

# Use same storage file as previous upload
storage = FileStorage('/tmp/tuspy_uploads.db')

# Create uploader for same file - will automatically resume
my_client = client.TusClient('http://tusd.tusdemo.net/files/')
uploader = my_client.uploader(
    '/path/to/large_file.ext',  # Same file path as before
    chunk_size=5*1024*1024,
    store_url=True,
    url_storage=storage
)

# Check if resuming from previous upload
if uploader.offset > 0:
    print(f"Resuming upload from offset {uploader.offset}")
    progress = (uploader.offset / uploader.get_file_size()) * 100
    print(f"Already uploaded: {progress:.1f}%")

try:
    uploader.upload()
    print("Upload completed")
finally:
    storage.close()

from tusclient.storage.interface import Storage
import redis

class RedisStorage(Storage):
    """Redis-based storage implementation."""
    
    def __init__(self, redis_client):
        self.redis = redis_client
    
    def get_item(self, key: str) -> Optional[str]:
        result = self.redis.get(f"tuspy:{key}")
        return result.decode('utf-8') if result else None
    
    def set_item(self, key: str, value: str):
        self.redis.set(f"tuspy:{key}", value)
    
    def remove_item(self, key: str):
        self.redis.delete(f"tuspy:{key}")

# Use custom storage
redis_client = redis.Redis(host='localhost', port=6379, db=0)
storage = RedisStorage(redis_client)

my_client = client.TusClient('http://tusd.tusdemo.net/files/')
uploader = my_client.uploader(
    '/path/to/file.ext',
    store_url=True,
    url_storage=storage
)

uploader.upload()

from tusclient.fingerprint.interface import Fingerprint
import hashlib
from typing import IO

class SHA256Fingerprint(Fingerprint):
    """SHA256-based fingerprint implementation."""
    
    def get_fingerprint(self, fs: IO) -> str:
        # Read entire small files, first 64KB of large files
        fs.seek(0)
        content = fs.read(65536)
        
        # Get file size
        fs.seek(0, 2)  # Seek to end
        size = fs.tell()
        
        # Generate SHA256 hash
        hasher = hashlib.sha256()
        hasher.update(content if isinstance(content, bytes) else content.encode('utf-8'))
        
        return f"size:{size}--sha256:{hasher.hexdigest()}"

# Use custom fingerprint
from tusclient.storage.filestorage import FileStorage

storage = FileStorage('/tmp/tuspy_uploads.db')
custom_fingerprint = SHA256Fingerprint()

my_client = client.TusClient('http://tusd.tusdemo.net/files/')
uploader = my_client.uploader(
    '/path/to/file.ext',
    store_url=True,
    url_storage=storage,
    fingerprinter=custom_fingerprint
)

uploader.upload()
storage.close()

from tusclient.storage.filestorage import FileStorage
from tusclient.fingerprint.fingerprint import Fingerprint

# Create storage and fingerprint instances
storage = FileStorage('/tmp/tuspy_uploads.db')
fingerprinter = Fingerprint()

# Generate fingerprint for a file
with open('/path/to/file.ext', 'rb') as fs:
    key = fingerprinter.get_fingerprint(fs)

# Check if URL is stored
stored_url = storage.get_item(key)
if stored_url:
    print(f"Found stored URL: {stored_url}")
else:
    print("No stored URL found for this file")

# Manually remove stored URL (e.g., after successful upload)
storage.remove_item(key)

storage.close()