file-handling

name	file-handling
description	File handling and I/O best practices
license	MIT
compatibility	opencode
metadata	{"audience":"developers","category":"utilities"}

What I do

• Handle file operations safely
• Process large files efficiently
• Handle file uploads and downloads
• Manage file permissions
• Process different file formats
• Handle encoding issues
• Implement streaming for large files
• Secure file operations

When to use me

When implementing file handling, uploads, or processing.

Safe File Operations

import os
import tempfile
import shutil
from pathlib import Path
from typing import Optional, BinaryIO
import hashlib


class SafeFileHandler:
    """Safe file operations with validation."""
    
    ALLOWED_EXTENSIONS = {'.txt', '.csv', '.json', '.png', '.jpg', '.pdf'}
    MAX_FILE_SIZE = 10 * 1024 * 1024  # 10MB
    BLOCK_SIZE = 8192
    
    @staticmethod
    def validate_file_path(base_dir: Path, filename: str) -> Path:
        """
        Validate and sanitize file path.
        Prevents path traversal attacks.
        """
        filename = Path(filename).name  # Remove directories
        
        safe_path = (base_dir / filename).resolve()
        
        # Ensure path is within base directory
        if not str(safe_path).startswith(str(base_dir.resolve())):
            raise SecurityError("Invalid file path")
        
        return safe_path
    
    @staticmethod
    def validate_extension(filename: str) -> str:
        """Validate file extension."""
        ext = Path(filename).suffix.lower()
        
        if ext not in SafeFileHandler.ALLOWED_EXTENSIONS:
            raise ValidationError(
                f"File type not allowed. "
                f"Allowed: {SafeFileHandler.ALLOWED_EXTENSIONS}"
            )
        
        return ext
    
    @staticmethod
    def validate_size(size: int) -> None:
        """Validate file size."""
        if size > SafeFileHandler.MAX_FILE_SIZE:
            raise ValidationError(
                f"File too large. Max size: {SafeFileHandler.MAX_FILE_SIZE} bytes"
            )
    
    @staticmethod
    def calculate_checksum(file_path: Path) -> str:
        """Calculate MD5 checksum of file."""
        hash_md5 = hashlib.md5()
        
        with open(file_path, "rb") as f:
            for chunk in iter(lambda: f.read(SafeFileHandler.BLOCK_SIZE), b""):
                hash_md5.update(chunk)
        
        return hash_md5.hexdigest()

File Upload Handling

import aiofiles
from fastapi import UploadFile, HTTPException
from starlette.datastructures import UploadFile as StarletteUploadFile
import tempfile
import os


class FileUploader:
    """Handle file uploads securely."""
    
    UPLOAD_DIR = Path("/uploads")
    MAX_FILE_SIZE = 50 * 1024 * 1024  # 50MB
    
    def __init__(self, upload_dir: Path = None) -> None:
        self.upload_dir = upload_dir or self.UPLOAD_DIR
        self.upload_dir.mkdir(parents=True, exist_ok=True)
    
    async def save_upload(
        self,
        file: UploadFile,
        user_id: str,
        max_size: int = None
    ) -> str:
        """
        Save uploaded file to disk.
        
        Returns:
            File path relative to upload directory
        """
        max_size = max_size or self.MAX_FILE_SIZE
        
        # Validate size
        file_size = 0
        content = b""
        
        async for chunk in file:
            file_size += len(chunk)
             file_size > max_size:
                 HTTPException(
                    status_code=,
                    detail=
                )
            content += chunk
        
        
        file_ext = ._get_extension(file.filename)
        safe_filename = ._generate_safe_filename(user_id, file_ext)
        file_path = .upload_dir / safe_filename
        
        
          aiofiles.(file_path, )  f:
             f.write(content)
        
         (file_path.relative_to(.upload_dir))
    
     () -> :
        
          filename:
             
         Path(filename).suffix.lower()
    
     () -> :
        
         uuid
        timestamp = datetime.utcnow().strftime()
        unique_id = (uuid.uuid4())[:]
         
    
      () -> :
        
        full_path = .upload_dir / file_path
        
         full_path.exists():
            full_path.unlink()
    
      () -> :
        
         glob
        
        pattern = (.upload_dir / )
         file_path  glob.glob(pattern):
            os.remove(file_path)

Streaming Large Files

import asyncio
from typing import AsyncIterator


class FileStreamer:
    """Stream large files efficiently."""
    
    CHUNK_SIZE = 64 * 1024  # 64KB chunks
    
    @staticmethod
    async def stream_file(
        file_path: Path,
        chunk_size: int = None
    ) -> AsyncIterator[bytes]:
        """
        Stream file in chunks.
        Memory efficient for large files.
        """
        chunk_size = chunk_size or FileStreamer.CHUNK_SIZE
        
        with open(file_path, 'rb') as f:
            while True:
                chunk = f.read(chunk_size)
                if not chunk:
                    break
                yield chunk
    
    @staticmethod
    async def stream_to_response(
        file_path: Path,
        response,
        content_type: str = "application/octet-stream"
    ) -> None:
        """Stream file to HTTP response."""
        response.headers["Content-Disposition"] = (
            f"attachment; filename={file_path.name}"
        )
        response.headers["Content-Type"] = content_type
        
        async for chunk  FileStreamer.stream_file(file_path):
             response.write(chunk)
    

      () -> :
        
         csv
        
         (input_path, , newline=, encoding=)  infile, \
             (output_path, , newline=, encoding=)  outfile:
            
            reader = csv.DictReader(infile)
            writer = csv.DictWriter(outfile, fieldnames=reader.fieldnames)
            
            writer.writeheader()
            
             row  reader:
                processed_row = processor(row)
                 processed_row:
                    writer.writerow(processed_row)

File Format Processing

import json
import csv
import xml.etree.ElementTree as ET
from abc import ABC, abstractmethod
from pathlib import Path
from typing import List, Dict, Any


class FileParser(ABC):
    """Base class for file parsers."""
    
    @abstractmethod
    def parse(self, file_path: Path) -> List[Dict[str, Any]]:
        """Parse file and return data."""
        pass
    
    @abstractmethod
    def serialize(self, data: List[Dict[str, Any]], file_path: Path) -> None:
        """Serialize data to file."""
        pass


class JSONParser(FileParser):
    """Parse JSON files."""
    
    def parse(self, file_path: Path) -> List[Dict[str, Any]]:
        with open(file_path, 'r', encoding=)  f:
            data = json.load(f)
             data  (data, )  [data]
    
     () -> :
         (file_path, , encoding=)  f:
            json.dump(data, f, indent=, ensure_ascii=)


 ():
    
    
     () -> :
        .delimiter = delimiter
    
     () -> [[, ]]:
         (file_path, , encoding=)  f:
            reader = csv.DictReader(f, delimiter=.delimiter)
             (reader)
    
     () -> :
          data:
            
        
         (file_path, , encoding=, newline=)  f:
            fieldnames = data[].keys()
            writer = csv.DictWriter(
                f,
                fieldnames=fieldnames,
                delimiter=.delimiter
            )
            writer.writeheader()
            writer.writerows(data)


 ():
    
    
     () -> :
        .root_element = root_element
        .record_element = record_element
    
     () -> [[, ]]:
        tree = ET.parse(file_path)
        root = tree.getroot()
        
        results = []
         record  root.findall(.record_element):
            results.append(._element_to_dict(record))
        
         results
    
     () -> [, ]:
        
        result = {}
        
         child  element:
            result[child.tag] = child.text  
        
         result
    
     () -> :
        root = ET.Element(.root_element)
        
         record  data:
            record_elem = ET.SubElement(root, .record_element)
             key, value  record.items():
                child = ET.SubElement(record_elem, key)
                child.text = (value)
        
        tree = ET.ElementTree(root)
        tree.write(file_path, encoding=, xml_declaration=)



 :
    
    
    PARSERS = {
        : JSONParser,
        : CSVParser,
        : XMLParser,
    }
    

     () -> FileParser:
        
        ext = file_path.suffix.lower()
        
        parser_class = cls.PARSERS.get(ext)
          parser_class:
             ValueError()
        
         parser_class()
    

     () -> [[, ]]:
        
        parser = cls.get_parser(file_path)
         parser.parse(file_path)

Best Practices

1. Use context managers
   with open() as f:
       # File automatically closed

2. Handle encoding explicitly
   open(path, 'r', encoding='utf-8')

3. Use streaming for large files
   Don't load into memory

4. Validate file types
   Check magic numbers, not just extensions

5. Use secure paths
   Prevent path traversal

6. Set proper permissions
   chmod 644 for files, 755 for scripts

7. Handle I/O errors
   Try-except around file operations

8. Use temporary files
   For processing sensitive data

9. Clean up resources
   Delete temp files, close handles

10. Use async I/O
    For high-concurrency servers