P2P Transport

from syft_flwr.client import SyftP2PClient

class SyftP2PClient(SyftFlwrClient):
    """Client for P2P (Google Drive) sync mode using Google Drive API.
    
    This client is used when FL jobs are submitted via syft_client's
    Google Drive-based sync system.
    
    Key differences from syft_core:
    - Uses Google Drive API directly (via GDriveFileIO) instead of filesystem paths
    - No RPC/crypto/event - Google Drive handles transport and access control
    - Email comes from environment variable or explicit parameter
    """

from syft_flwr.client import SyftP2PClient

client = SyftP2PClient(email="user@example.com")
print(client.email)  # user@example.com

@classmethod
def from_env(cls) -> SyftP2PClient

import os
os.environ["SYFTBOX_EMAIL"] = "user@example.com"

from syft_flwr.client import SyftP2PClient

client = SyftP2PClient.from_env()
print(client.email)  # user@example.com

@property
def email(self) -> str

@property
def syftbox_folder(self) -> Path

@property
def config_path(self) -> Path

@property
def my_datasite(self) -> Path

@property
def datasites(self) -> Path

def app_data(
    self,
    app_name: Optional[str] = None,
    datasite: Optional[str] = None,
) -> Path

client = SyftP2PClient(email="user@example.com")

# Get app data for current user
app_path = client.app_data("flwr/diabetes")
print(app_path)
# SyftBox/user@example.com/app_data/flwr/diabetes

# Get app data for another datasite
other_path = client.app_data("flwr/diabetes", "other@example.com")
print(other_path)
# SyftBox/other@example.com/app_data/flwr/diabetes

def get_client(self) -> SyftP2PClient

from syft_flwr.rpc import P2PFileRpc

class P2PFileRpc(SyftFlwrRpc):
    """P2P File-based RPC adapter using Google Drive API via syft-client.
    
    Instead of using filesystem paths, this adapter uses GDriveFileIO to:
    - Write .request files to the shared outbox folder via Google Drive API
    - Poll for .response files in the inbox folder via Google Drive API
    - Uses in-memory tracking for pending futures
    """

SyftBox/
├── syft_outbox_inbox_sender_to_recipient/
│   └── {app_name}/
│       └── rpc/
│           └── {endpoint}/
│               └── {future_id}.request
└── syft_outbox_inbox_recipient_to_sender/
    └── {app_name}/
        └── rpc/
            └── {endpoint}/
                └── {future_id}.response

def send(
    self,
    to_email: str,
    app_name: str,
    endpoint: str,
    body: bytes,
    encrypt: bool = False,
) -> str

from syft_flwr.rpc import P2PFileRpc

rpc = P2PFileRpc(
    sender_email="sender@example.com",
    app_name="flwr/my_app"
)

future_id = rpc.send(
    to_email="recipient@example.com",
    app_name="flwr/my_app",
    endpoint="messages",
    body=b"serialized_message",
    encrypt=False
)

print(f"Sent message, future_id: {future_id}")

def get_response(self, future_id: str) -> Optional[bytes]

import time

future_id = rpc.send(
    to_email="recipient@example.com",
    app_name="flwr/my_app",
    endpoint="messages",
    body=b"request_data"
)

# Poll for response
while True:
    response = rpc.get_response(future_id)
    if response is not None:
        print(f"Got response: {len(response)} bytes")
        break
    time.sleep(2)  # Wait before polling again

def delete_future(self, future_id: str) -> None

response = rpc.get_response(future_id)
if response is not None:
    process_response(response)
    rpc.delete_future(future_id)  # Clean up tracking

from syft_flwr.events import P2PFileEvents

class P2PFileEvents(SyftFlwrEvents):
    """P2P File-based polling events using Google Drive API via syft-client.
    
    This adapter:
    - Polls inbox folders for incoming .request files (from other participants) via Google Drive API
    - Calls the registered handler with the message bytes
    - Writes the response to the outbox folder (back to sender) via Google Drive API
    """

@property
def client_email(self) -> str

@property
def app_dir(self) -> Path

@property
def is_running(self) -> bool

def on_request(
    self,
    endpoint: str,
    handler: Callable[[bytes], Optional[Union[str, bytes]]],
    auto_decrypt: bool = True,
    encrypt_reply: bool = False,
) -> None

from syft_flwr.events import P2PFileEvents

events = P2PFileEvents(
    app_name="flwr/my_app",
    client_email="user@example.com",
    poll_interval=2.0
)

def handle_message(body: bytes) -> bytes:
    """Process incoming FL message."""
    print(f"Received: {len(body)} bytes")
    # Process message...
    return b"response_data"

events.on_request(
    endpoint="/messages",
    handler=handle_message
)

print("Starting polling loop...")
events.run_forever()

def run_forever(self) -> None

try:
    events.run_forever()
except KeyboardInterrupt:
    print("Shutting down...")
    events.stop()

def stop(self) -> None

import os
from syft_flwr.client import create_client
from syft_flwr.rpc import create_rpc
from syft_flwr.events import create_events_watcher

# Set environment for P2P mode
os.environ["SYFTBOX_EMAIL"] = "dataowner@example.com"

# Create client (auto-detects P2P from config)
client = create_client(
    transport="p2p",
    email="dataowner@example.com"
)

# Create RPC adapter (returns P2PFileRpc)
rpc = create_rpc(client=client, app_name="flwr/diabetes")

# Send a message
future_id = rpc.send(
    to_email="aggregator@example.com",
    app_name="flwr/diabetes",
    endpoint="messages",
    body=b"training_update"
)

# Create events watcher (returns P2PFileEvents)
events = create_events_watcher(
    app_name="flwr/diabetes",
    client=client,
    poll_interval=2.0
)

def handle_training_request(body: bytes) -> bytes:
    """Handle incoming training requests from aggregator."""
    # Train local model...
    return b"local_model_weights"

events.on_request("/messages", handle_training_request)
events.run_forever()