63 lines
2 KiB
Python
63 lines
2 KiB
Python
# Copyright (c) 2025 Patrick Motsch
|
|
# All rights reserved.
|
|
"""Abstract base classes for the Provider-Connector architecture (1:n).
|
|
|
|
One ProviderConnector per vendor (e.g. MsftConnector, GoogleConnector).
|
|
Each ProviderConnector exposes n ServiceAdapters (e.g. SharepointAdapter, OutlookAdapter).
|
|
All ServiceAdapters share the same access token from the UserConnection.
|
|
"""
|
|
|
|
from abc import ABC, abstractmethod
|
|
from dataclasses import dataclass, field
|
|
from typing import List, Optional, Union
|
|
|
|
|
|
@dataclass
|
|
class DownloadResult:
|
|
"""Rich return type for ServiceAdapter.download() when metadata is available."""
|
|
data: bytes = field(default=b"", repr=False)
|
|
fileName: str = ""
|
|
mimeType: str = ""
|
|
|
|
|
|
class ServiceAdapter(ABC):
|
|
"""Standardized operations for a single service of a provider."""
|
|
|
|
@abstractmethod
|
|
async def browse(self, path: str, filter: Optional[str] = None) -> list:
|
|
"""List items (files/folders) at the given path."""
|
|
...
|
|
|
|
@abstractmethod
|
|
async def download(self, path: str) -> Union[bytes, DownloadResult]:
|
|
"""Download a file. Return bytes or DownloadResult with metadata."""
|
|
...
|
|
|
|
@abstractmethod
|
|
async def upload(self, path: str, data: bytes, fileName: str) -> dict:
|
|
"""Upload a file to the given path. Returns metadata of the created entry."""
|
|
...
|
|
|
|
@abstractmethod
|
|
async def search(self, query: str, path: Optional[str] = None) -> list:
|
|
"""Search for items matching the query."""
|
|
...
|
|
|
|
|
|
class ProviderConnector(ABC):
|
|
"""One connector per provider. Manages a UserConnection + token.
|
|
Provides access to n services of the provider."""
|
|
|
|
def __init__(self, connection, accessToken: str):
|
|
self.connection = connection
|
|
self.accessToken = accessToken
|
|
|
|
@abstractmethod
|
|
def getAvailableServices(self) -> List[str]:
|
|
"""Which services does this provider offer?"""
|
|
...
|
|
|
|
@abstractmethod
|
|
def getServiceAdapter(self, service: str) -> ServiceAdapter:
|
|
"""Return the ServiceAdapter for a specific service."""
|
|
...
|