chore: refactor file upload mechanism to avoid NFS dependency

This commit is contained in:
Abhishek Kumar 2026-07-03 18:50:15 +05:30
parent 79a4a3c9f1
commit 6c5e1622f3
22 changed files with 293 additions and 400 deletions

View file

@ -1,23 +1,51 @@
from abc import ABC, abstractmethod
from typing import Any, BinaryIO, Dict, Optional
from typing import Any, Dict, Optional, Protocol
class AsyncReadable(Protocol):
"""Anything exposing ``await .read() -> bytes`` (aiofiles handles, in-memory wrappers)."""
async def read(self) -> bytes: ...
class _AsyncBytesReader:
"""Async file-like wrapper over in-memory bytes for acreate_file()."""
def __init__(self, data: bytes):
self._data = data
async def read(self) -> bytes:
return self._data
class BaseFileSystem(ABC):
"""Abstract base class for filesystem operations."""
@abstractmethod
async def acreate_file(self, file_path: str, content: BinaryIO) -> bool:
async def acreate_file(self, file_path: str, content: AsyncReadable) -> bool:
"""Create a new file with the given content.
Args:
file_path: Path where the file should be created
content: File content as a binary stream
content: File content readable via ``await content.read()``
Returns:
bool: True if file was created successfully, False otherwise
"""
pass
async def acreate_file_from_bytes(self, file_path: str, data: bytes) -> bool:
"""Create a file directly from in-memory bytes (no local file needed).
Args:
file_path: Path where the file should be created
data: File content as bytes
Returns:
bool: True if file was created successfully, False otherwise
"""
return await self.acreate_file(file_path, _AsyncBytesReader(data))
@abstractmethod
async def aupload_file(self, local_path: str, destination_path: str) -> bool:
"""Upload a file from local path to destination.

View file

@ -1,11 +1,11 @@
import asyncio
import os
from datetime import datetime
from typing import BinaryIO, Optional
from typing import Optional
import aiofiles
from .base import BaseFileSystem
from .base import AsyncReadable, BaseFileSystem
class LocalFileSystem(BaseFileSystem):
@ -24,7 +24,7 @@ class LocalFileSystem(BaseFileSystem):
"""Get the full path by joining with base path."""
return os.path.join(self.base_path, file_path)
async def acreate_file(self, file_path: str, content: BinaryIO) -> bool:
async def acreate_file(self, file_path: str, content: AsyncReadable) -> bool:
try:
full_path = self._get_full_path(file_path)
os.makedirs(os.path.dirname(full_path), exist_ok=True)

View file

@ -1,12 +1,12 @@
import asyncio
import json
from typing import Any, BinaryIO, Dict, Optional
from typing import Any, Dict, Optional
from loguru import logger
from minio import Minio
from minio.error import S3Error
from .base import BaseFileSystem
from .base import AsyncReadable, BaseFileSystem
class MinioFileSystem(BaseFileSystem):
@ -89,7 +89,7 @@ class MinioFileSystem(BaseFileSystem):
logger.debug(f"Bucket setup note: {e}")
pass
async def acreate_file(self, file_path: str, content: BinaryIO) -> bool:
async def acreate_file(self, file_path: str, content: AsyncReadable) -> bool:
try:
data = await content.read()

View file

@ -1,6 +1,6 @@
from typing import Any, BinaryIO, Dict, NoReturn, Optional
from typing import Any, Dict, NoReturn, Optional
from .base import BaseFileSystem
from .base import AsyncReadable, BaseFileSystem
class NullFileSystem(BaseFileSystem):
@ -16,7 +16,7 @@ class NullFileSystem(BaseFileSystem):
"Set ENVIRONMENT to a non-test value or inject a real filesystem fixture."
)
async def acreate_file(self, file_path: str, content: BinaryIO) -> bool:
async def acreate_file(self, file_path: str, content: AsyncReadable) -> bool:
self._fail("acreate_file")
async def aupload_file(self, local_path: str, destination_path: str) -> bool:

View file

@ -1,10 +1,10 @@
from typing import Any, BinaryIO, Dict, Optional
from typing import Any, Dict, Optional
import aioboto3
from botocore.config import Config
from botocore.exceptions import ClientError
from .base import BaseFileSystem
from .base import AsyncReadable, BaseFileSystem
class S3FileSystem(BaseFileSystem):
@ -57,7 +57,7 @@ class S3FileSystem(BaseFileSystem):
kwargs["config"] = self._config
return kwargs
async def acreate_file(self, file_path: str, content: BinaryIO) -> bool:
async def acreate_file(self, file_path: str, content: AsyncReadable) -> bool:
try:
async with self.session.client("s3", **self._client_kwargs()) as s3_client:
await s3_client.put_object(