Chapter 21: File Uploads and Binary Data

# Text data - what you've been doing
text_data = {"name": "Alice", "age": 25}

# requests converts this dictionary to a UTF-8 string automatically
response = requests.post("https://httpbin.org/post", json=text_data)

# WRONG - This corrupts binary files
with open('profile.jpg', 'r') as f:  # Text mode tries to decode bytes
    data = f.read()  # ❌ Decoding error or corrupted data

# CORRECT - Binary mode preserves raw bytes
with open('profile.jpg', 'rb') as f:  # Binary mode reads raw bytes
    data = f.read()  # ✅ Data preserved exactly as-is
    print(f"First 4 bytes: {data[:4]}")  # Shows actual hex values

First 4 bytes: b'\xff\xd8\xff\xe0'

import requests

# 1. Create a test file
with open('test.txt', 'w') as f:
    f.write("This is a test file for upload.")

# 2. Open in binary mode and upload
with open('test.txt', 'rb') as f:
    files = {'file': f}
    response = requests.post("https://httpbin.org/post", files=files, timeout=10)

# 3. Inspect the response
print(f"Status: {response.status_code}")
print(f"Content-Type sent: {response.request.headers.get('Content-Type')}")
print(f"File received by server: {response.json()['files']}")

Status: 200
Content-Type sent: multipart/form-data; boundary=...
File received by server: {'file': 'This is a test file for upload.'}

import requests

# Note: You need an actual image file for this
# For testing, create a small dummy binary file:
with open('profile.jpg', 'wb') as f:
    f.write(b'\xFF\xD8\xFF\xE0')  # JPEG header bytes

# Upload with metadata
with open('profile.jpg', 'rb') as f:
    files = {'profile_image': f}
    data = {
        'user_id': '12345',
        'description': 'Profile photo taken at conference'
    }
    
    response = requests.post(
        "https://httpbin.org/post", 
        files=files,
        data=data,  # Metadata goes in 'data', not 'json'
        timeout=15
    )

result = response.json()
print(f"Files: {result['files']}")
print(f"Form data: {result['form']}")

Files: {'profile_image': ...}
Form data: {'user_id': '12345', 'description': 'Profile photo taken at conference'}

import os
from pathlib import Path
import mimetypes
import requests

class ProfilePictureUploader:
    """
    A production-style profile picture uploader with:
      - Extension and size limits
      - Basic header validation (magic bytes)
      - Safe filenames
      - Proper MIME types
      - Clear (success, message, url) return structure
    """

    # Maximum file size: 5MB
    MAX_FILE_SIZE = 5 * 1024 * 1024

    # Allowed image extensions
    ALLOWED_EXTENSIONS = {'.jpg', '.jpeg', '.png', '.gif'}

    def __init__(self, user_id):
        self.user_id = user_id

    def upload_profile_picture(self, filepath):
        """
        Upload a profile picture with comprehensive validation.

        Returns:
            tuple: (success: bool, message: str, url: str or None)
        """

        # STEP 1: File existence check
        if not os.path.exists(filepath):
            return (False, "File not found", None)

        # STEP 2: Extension validation
        file_ext = Path(filepath).suffix.lower()
        if file_ext not in self.ALLOWED_EXTENSIONS:
            allowed = ', '.join(self.ALLOWED_EXTENSIONS)
            return (False, f"Invalid file type. Allowed: {allowed}", None)

        # STEP 3: File size validation
        file_size = os.path.getsize(filepath)
        if file_size > self.MAX_FILE_SIZE:
            max_mb = self.MAX_FILE_SIZE / (1024 * 1024)
            actual_mb = file_size / (1024 * 1024)
            return (False, f"File too large ({actual_mb:.1f}MB). Max: {max_mb}MB", None)

        # STEP 4: Basic content validation (is it really an image?)
        try:
            with open(filepath, 'rb') as f:
                header = f.read(10)

            if not self._is_valid_image_header(header, file_ext):
                return (False, "File content does not match a valid image format", None)
        except Exception as e:
            return (False, f"Failed to read file header: {str(e)}", None)

        # STEP 5: Prepare safe filename
        safe_filename = self._build_safe_filename(file_ext)

        # STEP 6: Determine correct MIME type
        mime_type = self._get_mime_type(file_ext)

        # STEP 7: Upload to remote API
        try:
            with open(filepath, 'rb') as f:
                files = {
                    'file': (safe_filename, f, mime_type)
                }

                response = requests.post(
                    "https://httpbin.org/post",
                    files=files,
                    timeout=30
                )
                response.raise_for_status()

                # In a real system, you'd parse the response to get the final URL.
                # Here we just return a pretend URL.
                uploaded_url = f"https://cdn.example.com/profile_pics/{safe_filename}"
                return (True, "Upload successful", uploaded_url)

        except requests.exceptions.Timeout:
            return (False, "Upload timed out - please try again", None)

        except requests.exceptions.RequestException as e:
            return (False, f"Upload failed: {str(e)}", None)

    def _is_valid_image_header(self, header, extension):
        """Check if file header matches the claimed extension"""
        # JPEG: FF D8 FF
        if extension in {'.jpg', '.jpeg'} and header[:3] == b'\xFF\xD8\xFF':
            return True
        # PNG: 89 50 4E 47 0D 0A 1A 0A
        if extension == '.png' and header[:8] == b'\x89PNG\r\n\x1a\n':
            return True
        # GIF: 47 49 46 38 37 61 or 47 49 46 38 39 61
        if extension == '.gif' and header[:6] in (b'GIF87a', b'GIF89a'):
            return True
        return False

    def _get_mime_type(self, extension):
        """Get proper MIME type for extension"""
        mime_types = {
            '.jpg': 'image/jpeg',
            '.jpeg': 'image/jpeg',
            '.png': 'image/png',
            '.gif': 'image/gif'
        }
        return mime_types.get(extension, 'application/octet-stream')

    def _build_safe_filename(self, extension):
        """
        Build a safe filename that does not leak original names
        (which might contain personal data).
        """
        return f"user_{self.user_id}_profile{extension}"


# Usage Example
if __name__ == "__main__":
    uploader = ProfilePictureUploader(user_id=12345)

    # In a real app, this would be a file path chosen by the user.
    test_image = "test_profile_picture.jpg"

    success, message, url = uploader.upload_profile_picture(test_image)
    print("Success:", success)
    print("Message:", message)
    print("URL:", url)

import os
import requests

class ProgressFileReader:
    """Generator that reads a file in chunks and displays progress."""
    
    def __init__(self, filepath, chunk_size=8192):
        self.filepath = filepath
        self.chunk_size = chunk_size
        self.file_size = os.path.getsize(filepath)
        self.bytes_read = 0
    
    def __iter__(self):
        with open(self.filepath, "rb") as f:
            while True:
                chunk = f.read(self.chunk_size)
                if not chunk:
                    break
                
                self.bytes_read += len(chunk)
                percentage = (self.bytes_read / self.file_size) * 100
                print(f"\rUploading: {percentage:.1f}% ({self.bytes_read}/{self.file_size} bytes)", end="")
                
                yield chunk
        print()  # New line after completion

# Create a test file (5MB)
with open("large_test.dat", "wb") as f:
    f.write(os.urandom(5 * 1024 * 1024))

print("Starting upload...")
reader = ProgressFileReader("large_test.dat")

# We stream the data directly using the reader object
response = requests.post(
    "https://httpbin.org/post",
    data=reader,
    timeout=300  # Longer timeout for large files
)

print("\nUpload Complete!")

Starting upload...
Uploading: 45.2% (2367488/5242880 bytes)
...
Uploading: 100.0% (5242880/5242880 bytes)

Upload Complete!

import requests

def download_file(url, output_path, chunk_size=8192):
    """
    Downloads a file with streaming and content-type validation.
    """
    print(f"Downloading from {url}...")
    
    # stream=True prevents loading the whole response into memory
    with requests.get(url, stream=True, timeout=30) as response:
        response.raise_for_status()
        
        # Validation: Ensure it's actually an image (or expected type)
        content_type = response.headers.get('Content-Type', '')
        if 'image' not in content_type:
            print(f"Warning: Content-Type is '{content_type}', expected an image.")
        
        # Save to file
        with open(output_path, 'wb') as f:
            for chunk in response.iter_content(chunk_size=chunk_size):
                f.write(chunk)
                
    print(f"Saved to {output_path}")

# Test it
download_file("https://httpbin.org/image/jpeg", "downloaded_image.jpg")

import requests
from concurrent.futures import ThreadPoolExecutor

def upload_single_file(filename):
    """Helper function to upload one file"""
    try:
        with open(filename, 'rb') as f:
            response = requests.post(
                "https://httpbin.org/post", 
                files={'file': f},
                timeout=30
            )
        return f"✅ {filename}: {response.status_code}"
    except Exception as e:
        return f"❌ {filename}: {str(e)}"

files_to_upload = ['doc1.txt', 'doc2.txt', 'doc3.txt']

# Create dummy files for the test
for f in files_to_upload:
    with open(f, 'w') as file: 
        file.write("content")

print("Starting concurrent upload...")

# Upload up to 3 files at once
with ThreadPoolExecutor(max_workers=3) as executor:
    results = list(executor.map(upload_single_file, files_to_upload))

for result in results:
    print(result)

Starting concurrent upload...
✅ doc1.txt: 200
✅ doc2.txt: 200
✅ doc3.txt: 200

import requests
import re
import os

class ReceiptScanner:
    def __init__(self, api_key=None):
        """
        api_key:
            Your OCR.space API key. When omitted, this falls back to the
            OCRSPACE_API_KEY environment variable or 'helloworld' for the demo key.
        """
        if api_key is None:
            api_key = os.environ.get("OCRSPACE_API_KEY", "helloworld")
        self.api_key = api_key
        self.url = "https://api.ocr.space/parse/image"

    def scan_receipt(self, image_path):
        print(f"Scanning {image_path}...")

        # 1. Validate File
        if not os.path.exists(image_path):
            return {"error": "File not found"}

        # 2. Upload and Process (Synchronous)
        try:
            with open(image_path, "rb") as f:
                payload = {
                    "apikey": self.api_key,
                    "language": "eng",
                    "isOverlayRequired": False,
                }
                files = {"file": f}

                response = requests.post(
                    self.url,
                    files=files,
                    data=payload,
                    timeout=60,  # OCR takes time
                )
                response.raise_for_status()

                result = response.json()

                # Check for API-level errors
                if result.get("IsErroredOnProcessing"):
                    error_message = result.get("ErrorMessage")
                    if isinstance(error_message, list):
                        error_message = "; ".join(error_message)
                    return {"error": error_message or "Unknown OCR error"}

                # 3. Extract Text
                parsed_results = result.get("ParsedResults", [])
                if not parsed_results:
                    return {"error": "No text found"}

                raw_text = parsed_results[0].get("ParsedText", "")
                return self._parse_data(raw_text)

        except Exception as e:
            return {"error": str(e)}

    def _parse_data(self, text):
        """Extract amount and date using regex."""
        print("Analyzing text...")

        # Regex for currency (for example, $12.99 or 12.99)
        # Looks for lines starting with Total or Amount
        amount_match = re.search(
            r"(Total|Amount)[:\s]*\$?([\d,]+\.\d{2})",
            text,
            flags=re.IGNORECASE,
        )

        # Regex for date in common formats (dd/mm/yyyy, mm/dd/yyyy, yyyy-mm-dd)
        date_match = re.search(
            r"(\d{1,2}[\/\-]\d{1,2}[\/\-]\d{2,4}|\d{4}[\/\-]\d{1,2}[\/\-]\d{1,2})",
            text,
        )

        data = {}
        if amount_match:
            data["total_amount"] = amount_match.group(2)
        if date_match:
            data["date"] = date_match.group(1)

        if not data:
            data["note"] = "No amount or date found. Check OCR quality."

        return data

if __name__ == "__main__":
    scanner = ReceiptScanner()  # Uses env var or demo key by default
    result = scanner.scan_receipt("sample_receipt.jpg")
    print(result)