DEPS Object Storage

A unified Python library for interacting with cloud object storage services (AWS S3, Azure Blob Storage, Google Cloud Storage) with support for both synchronous and asynchronous operations.

Features

• Multi-cloud support: AWS S3, Azure Blob Storage, Google Cloud Storage
• Dual mode operation: Synchronous and asynchronous APIs
• Type-safe: Full type hints and mypy support
• Factory pattern: Easy instantiation based on configuration
• Dockerized development: Complete containerized development environment

Installation

Prerequisites

• Python 3.11+
• Docker and Docker Compose
• Poetry (for dependency management)

Setup

1. Clone the repository:

git clone <repository-url>
cd deps-object-storage

2. Install dependencies and setup environment:

make install
make prereq

3. Build the Docker container:

make build

Configuration

The library uses environment variables for configuration. Create a .env file with the following variables:

# Object Storage Configuration
OBJECT_STORAGE_TYPE=aws  # Options: aws, azure, gcp
OBJECT_STORAGE_MODE=async  # Options: sync, async

# Cloud-specific credentials (set based on your chosen provider)
# AWS
AWS_ACCESS_KEY_ID=your_access_key
AWS_SECRET_ACCESS_KEY=your_secret_key
AWS_DEFAULT_REGION=us-east-1

# Azure
AZURE_STORAGE_CONNECTION_STRING=your_connection_string

# Google Cloud
GOOGLE_APPLICATION_CREDENTIALS=path/to/service-account.json

Usage

Basic Usage

from deps_object_storage import make_object_storage

# Create object storage instance based on environment configuration
storage = make_object_storage()

# Upload a file
content = b"Hello, World!"
path = "example/file.txt"
uploaded_path = storage.upload(path, content, replace_if_exists=True)

# Download a file
downloaded_content = storage.download(path)

# Delete a file
storage.delete(path)

Asynchronous Usage

import asyncio
from deps_object_storage import make_object_storage

async def main():
    storage = make_object_storage()

    if hasattr(storage, 'upload'):  # Sync storage
        path = storage.upload("async_example.txt", b"Async content", True)
    else:  # Async storage
        path = await storage.upload("async_example.txt", b"Async content", True)
        content = await storage.download(path)
        await storage.delete(path)

asyncio.run(main())

API Reference

ObjectStorage Protocol

class ObjectStorage(Protocol):
    def upload(self, path: str, content: bytes, replace_if_exists: bool) -> str:
        """Upload content to the specified path."""
        ...

    def download(self, path: str) -> bytes:
        """Download content from the specified path."""
        ...

    def delete(self, path: str) -> None:
        """Delete the file at the specified path."""
        ...

AsyncObjectStorage Protocol

class AsyncObjectStorage(Protocol):
    async def upload(self, path: str, content: bytes, replace_if_exists: bool) -> str:
        """Upload content to the specified path asynchronously."""
        ...

    async def download(self, path: str) -> bytes:
        """Download content from the specified path asynchronously."""
        ...

    async def delete(self, path: str) -> None:
        """Delete the file at the specified path asynchronously."""
        ...

Development

Available Make Commands

• make install - Copy environment variables from .env.example
• make prereq - Create Docker network and ensure .env exists
• make build - Build Docker container
• make format - Format code using black and isort
• make format-check - Check code formatting without making changes
• make lint - Run flake8 linter
• make mypy-check - Run type checking with mypy
• make ci - Run all CI checks (format, lint, type check)
• make clean - Clean Python cache files

Development Workflow

1. Start development environment:

make build

2. Format code:

make format

3. Run code quality checks:

make ci

Project Structure

deps_object_storage/
├── __init__.py                 # Main package exports
├── object_storage.py          # Synchronous storage protocol
├── async_object_storage.py    # Asynchronous storage protocol
├── object_storage_factory.py  # Factory for creating storage instances
├── settings.py               # Configuration management
├── aws/                      # AWS S3 implementation
├── azure/                    # Azure Blob Storage implementation
├── gcp/                      # Google Cloud Storage implementation
└── file_storage/             # DEPS File Storage implementation

Dependencies

Runtime Dependencies

• pydantic-settings - Configuration management
• boto3 - AWS SDK for Python
• azure-storage-blob - Azure Blob Storage client
• google-cloud-storage - Google Cloud Storage client
• google-cloud-core - Google Cloud Core library
• asyncer - Async utilities
• aiohttp - Async HTTP

Development Dependencies

• mypy - Type checking
• flake8 - Linting
• isort - Import sorting
• black - Code formatting

License

MIT License - see the LICENSE file for details.

Contributing

1. Fork the repository
2. Create a feature branch
3. Make your changes
4. Run make ci to ensure code quality
5. Submit a pull request

Author

Valentin