streamlift-megapy

A modern Python client for the MEGA cloud storage service, built for streaming uploads with zero intermediate disk usage.

This package is designed as a drop-in replacement for the unmaintained mega.py package (with upload_stream support), and reimplements the core MEGA client protocol that the megajs npm package uses.

Features

• Streaming uploads — upload from any file-like object (or async generator) directly to MEGA without writing to local disk.
• Session restore — log in once, serialize the session, and resume later without a password (useful for long-running workers in ephemeral environments like Google Colab).
• Sync + async — both a blocking Mega class and an AsyncMega class backed by aiohttp.
• File management — create folders, delete, move, rename, and generate public share links.
• Downloads — download public files or logged-in files to disk.
• Quota — check account storage space and transfer quota.

Installation

pip install -e /path/to/streamlift-megapy
# or, once published:
pip install git+https://github.com/your-org/streamlift-megapy.git

Quick Start

Sync

from streamlift_megapy import Mega

mega = Mega().login("you@example.com", "password")

# Upload a file-like object (zero disk)
with open("movie.mp4", "rb") as f:
    file_node = mega.upload_stream(f, 123456789, "movie.mp4")

# Or upload by path
node = mega.upload("photo.jpg")

# Create + find folders
mega.create_folder("MyFolder")
folder = mega.find("MyFolder")

# Download
mega.download(node, destination="./out/")

# Share link
print(node.link())

Async

import asyncio
from streamlift_megapy import AsyncMega

async def main():
    async with AsyncMega().login("you@example.com", "password") as mega:
        node = await mega.upload_stream(file_obj, file_size, "movie.mp4")

asyncio.run(main())

Session restore

mega = Mega().login("you@example.com", "password")
session_json = mega.to_json()      # safe to persist
sid = mega.sid                     # string SID

# Later, without a password:
mega2 = Mega().login(session=session_json)
# or pass the raw SID string:
mega3 = Mega().login(session=sid)

API Overview

Development

pip install -e ".[dev]"
pytest
ruff check mega

License

MIT