Python SDK

The plugport Python package provides a PyMongo-compatible API. If you've used PyMongo, you already know this SDK.

Installation

pip install plugport

Quick Start

from plugport import PlugPortClient

client = PlugPortClient("http://localhost:8080")
db = client["myapp"]
users = db["users"]

# Insert
result = users.insert_one({"name": "Alice", "email": "alice@example.com", "age": 30})
print(f"Inserted: {result.inserted_id}")

# Find
docs = users.find({"age": {"$gte": 25}})
for doc in docs:
    print(doc)

client.close()

PyMongo Compatibility

The SDK mirrors PyMongo's API patterns:

# Dict-style access (just like PyMongo)
db = client["mydb"]
collection = db["users"]

# Attribute-style access
db = client.mydb
collection = db.users

# Context manager
with PlugPortClient("http://localhost:8080") as client:
    db = client["myapp"]
    # Auto-closes on exit

API Reference

PlugPortClient(uri, api_key=None)

# Basic connection
client = PlugPortClient("http://localhost:8080")

# With API key
client = PlugPortClient("http://localhost:8080", api_key="your-key")

# Using plugport:// scheme
client = PlugPortClient("plugport://localhost:8080")

Methods

Method	Returns	Description
client["name"]	Database	Get database by name
client.get_database(name)	Database	Get database by name
client.server_info()	dict	Server health info
client.list_database_names()	list[str]	List databases
client.close()	None	Close connection

Database

db = client["myapp"]

# List collections
names = db.list_collection_names()

# Drop a collection
db.drop_collection("users")

Collection

insert_one(document) -> InsertOneResult

result = users.insert_one({"name": "Alice", "email": "alice@example.com"})
print(result.acknowledged)  # True
print(result.inserted_id)   # "67b2a1f0..."

insert_many(documents) -> InsertManyResult

result = users.insert_many([
    {"name": "Alice", "age": 30},
    {"name": "Bob", "age": 25},
])
print(result.inserted_count)  # 2
print(result.inserted_ids)    # ["...", "..."]

find(filter=None, projection=None, sort=None, limit=0, skip=0) -> list[dict]

# All documents
docs = users.find()

# With filter
docs = users.find({"age": {"$gte": 18}})

# With all options
docs = users.find(
    filter={"status": "active"},
    projection={"name": 1, "score": 1},
    sort={"score": -1},
    limit=10,
    skip=0,
)

find_one(filter=None) -> dict | None

user = users.find_one({"email": "alice@example.com"})
if user:
    print(user["name"])

aggregate(pipeline) -> list

Execute an aggregation pipeline. Supported stages: $match, $lookup, $project, $sort, $limit, $skip, $unwind, $count.

results = orders.aggregate([
    {"$match": {"status": "completed"}},
    {"$lookup": {
        "from": "users",
        "localField": "userId",
        "foreignField": "_id",
        "as": "user",
    }},
    {"$unwind": "$user"},
    {"$sort": {"total": -1}},
    {"$limit": 10},
])
for order in results:
    print(f"{order['orderId']}: {order['user']['name']}")

update_one(filter, update, upsert=False) -> UpdateResult

result = users.update_one(
    {"name": "Alice"},
    {"$set": {"age": 31}},
    upsert=False,
)
print(result.matched_count)   # 1
print(result.modified_count)  # 1

delete_one(filter) -> DeleteResult

result = users.delete_one({"name": "Alice"})
print(result.deleted_count)  # 1

delete_many(filter) -> DeleteResult

result = users.delete_many({"status": "inactive"})
print(result.deleted_count)  # 5

create_index(field, unique=False) -> str

index_name = users.create_index("email", unique=True)
# "email_1"

count_documents(filter=None) -> int

count = users.count_documents({"status": "active"})

grant_role(address, role) -> dict

Grant an access role to a wallet address on this collection.

users.grant_role("0xabc...", 1)  # 1 = read, 2 = write

revoke_role(address) -> dict

Revoke access for a wallet address.

users.revoke_role("0xabc...")

Error Handling

from plugport.errors import PlugPortError, DuplicateKeyError, ConnectionError

try:
    users.insert_one({"email": "alice@example.com"})
except DuplicateKeyError as e:
    print(f"Duplicate: {e.message}")  # code: 11000
except ConnectionError as e:
    print(f"Connection failed: {e.message}")
except PlugPortError as e:
    print(f"Error [{e.code}]: {e.message}")

Migration from PyMongo

- from pymongo import MongoClient
+ from plugport import PlugPortClient

- client = MongoClient("mongodb://localhost:27017")
+ client = PlugPortClient("http://localhost:8080")

  # All code below is identical
  db = client["myapp"]
  users = db["users"]
  users.insert_one({"name": "Alice"})
  users.find({"age": {"$gte": 25}})