Add File Station support

pyproject.toml

@@ -73,7 +73,9 @@ disable = [
     "too-many-arguments",
     "too-many-branches",
     "too-many-instance-attributes",
+    "too-many-locals",
     "too-many-public-methods",
-    "too-many-return-statements",
     "too-many-positional-arguments",
+    "too-many-statements",
+    "too-many-return-statements",
 ]

src/synology_dsm/api/file_station/__init__.py

@@ -0,0 +1,125 @@
+"""Synology FileStation API wrapper."""
+
+from __future__ import annotations
+
+from collections.abc import AsyncIterator
+from io import BufferedReader
+
+from synology_dsm.api import SynoBaseApi
+
+from .models import (
+    SynoFileAdditionalOwner,
+    SynoFileAdditionalPermission,
+    SynoFileAdditionalTime,
+    SynoFileAdditionalVolumeStatus,
+    SynoFileFile,
+    SynoFileFileAdditional,
+    SynoFileSharedFolder,
+    SynoFileSharedFolderAdditional,
+)
+
+
+class SynoFileStation(SynoBaseApi):
+    """An implementation of a Synology FileStation."""
+
+    API_KEY = "SYNO.FileStation.*"
+    LIST_API_KEY = "SYNO.FileStation.List"
+    DOWNLOAD_API_KEY = "SYNO.FileStation.Download"
+    UPLOAD_API_KEY = "SYNO.FileStation.Upload"
+
+    async def get_shared_folders(
+        self, offset: int = 0, limit: int = 100, only_writable: bool = False
+    ) -> list[SynoFileSharedFolder] | None:
+        """Get a list of all shared folders."""
+        raw_data = await self._dsm.get(
+            self.LIST_API_KEY,
+            "list_share",
+            {
+                "offset": offset,
+                "limit": limit,
+                "onlywritable": only_writable,
+                "additional": (
+                    '["real_path","owner","time","perm",'
+                    '"mount_point_type","sync_share","volume_status"]'
+                ),
+            },
+        )
+        if not isinstance(raw_data, dict) or (data := raw_data.get("data")) is None:
+            return None
+
+        shared_folders: list[SynoFileSharedFolder] = []
+        for folder in data["shares"]:
+            additional = folder["additional"]
+            shared_folders.append(
+                SynoFileSharedFolder(
+                    SynoFileSharedFolderAdditional(
+                        additional["mount_point_type"],
+                        SynoFileAdditionalOwner(**additional["owner"]),
+                        SynoFileAdditionalPermission(**additional["perm"]),
+                        SynoFileAdditionalVolumeStatus(
+                            **additional["volume_status"],
+                        ),
+                    ),
+                    folder["isdir"],
+                    folder["name"],
+                    folder["path"],
+                )
+            )
+
+        return shared_folders
+
+    async def get_files(
+        self, path: str, offset: int = 0, limit: int = 100
+    ) -> list[SynoFileFile] | None:
+        """Get a list of all files in a folder."""
+        raw_data = await self._dsm.get(
+            self.LIST_API_KEY,
+            "list",
+            {
+                "offset": offset,
+                "limit": limit,
+                "folder_path": path,
+                "additional": (
+                    '["real_path","owner","time","perm",'
+                    '"mount_point_type","type","size"]'
+                ),
+            },
+        )
+        if not isinstance(raw_data, dict) or (data := raw_data.get("data")) is None:
+            return None
+
+        files: list[SynoFileFile] = []
+        for file in data["files"]:
+            additional = file["additional"]
+            files.append(
+                SynoFileFile(
+                    SynoFileFileAdditional(
+                        additional["mount_point_type"],
+                        SynoFileAdditionalOwner(**additional["owner"]),
+                        SynoFileAdditionalPermission(**additional["perm"]),
+                        additional["real_path"],
+                        additional["size"],
+                        SynoFileAdditionalTime(**additional["time"]),
+                        additional["type"],
+                    ),
+                    file["isdir"],
+                    file["name"],
+                    file["path"],
+                )
+            )
+
+        return files
+
+    async def upload_files(
+        self,
+        path: str,
+        filename: str,
+        content: bytes | BufferedReader | AsyncIterator[bytes],
+    ) -> bool | None:
+        """Upload a file to a folder."""
+        raw_data = await self._dsm.post_upload(
+            self.UPLOAD_API_KEY, "upload", path, filename, content
+        )
+        if not isinstance(raw_data, dict):
+            return None
+        return raw_data.get("success")

src/synology_dsm/api/file_station/const.py

@@ -0,0 +1 @@
+"""Synology FileStation API constants."""

src/synology_dsm/api/file_station/models.py

@@ -0,0 +1,100 @@
+"""Data models for Synology FileStation Module."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+# -------------------------------------
+# generic additional data
+# -------------------------------------
+
+
+@dataclass
+class SynoFileAdditionalOwner:
+    """Representation of an Synology FileStation additionl owner data."""
+
+    gid: int
+    group: str
+    uid: int
+    user: str
+
+
+@dataclass
+class SynoFileAdditionalPermission:
+    """Representation of an Synology FileStation additionl permission data."""
+
+    acl: dict
+    is_acl_mode: bool
+    posix: int
+
+
+# -------------------------------------
+# shared folder
+# -------------------------------------
+
+
+@dataclass
+class SynoFileAdditionalVolumeStatus:
+    """Representation of an Synology FileStation additionl permission data."""
+
+    freespace: int
+    totalspace: int
+    readonly: bool
+
+
+@dataclass
+class SynoFileSharedFolderAdditional:
+    """Representation of an Synology FileStation Shared Folder additionl data."""
+
+    mount_point_type: str
+    owner: SynoFileAdditionalOwner
+    perm: SynoFileAdditionalPermission
+    volume_status: SynoFileAdditionalVolumeStatus
+
+
+@dataclass
+class SynoFileSharedFolder:
+    """Representation of an Synology FileStation Shared Folder."""
+
+    addidtionan: SynoFileSharedFolderAdditional
+    is_dir: bool
+    name: str
+    path: str
+
+
+# -------------------------------------
+# file
+# -------------------------------------
+
+
+@dataclass
+class SynoFileAdditionalTime:
+    """Representation of an Synology FileStation additionl permission data."""
+
+    atime: int
+    ctime: int
+    crtime: int
+    mtime: int
+
+
+@dataclass
+class SynoFileFileAdditional:
+    """Representation of an Synology FileStation File additionl data."""
+
+    mount_point_type: str
+    owner: SynoFileAdditionalOwner
+    perm: SynoFileAdditionalPermission
+    real_path: str
+    size: int
+    time: SynoFileAdditionalTime
+    type: str
+
+
+@dataclass
+class SynoFileFile:
+    """Representation of an Synology FileStation File."""
+
+    addidtionan: SynoFileFileAdditional
+    is_dir: bool
+    name: str
+    path: str

src/synology_dsm/synology_dsm.py

@@ -5,12 +5,15 @@
 import asyncio
 import logging
 import socket
+from collections.abc import AsyncIterator
+from hashlib import md5
+from io import BufferedReader
 from ipaddress import IPv6Address
 from json import JSONDecodeError
-from typing import Any, Coroutine, TypedDict
+from typing import TYPE_CHECKING, Any, Coroutine, TypedDict
 from urllib.parse import quote, urlencode
 
-from aiohttp import ClientError, ClientSession, ClientTimeout
+from aiohttp import ClientError, ClientSession, ClientTimeout, MultipartWriter, hdrs
 from yarl import URL
 
 from .api import SynoBaseApi
@@ -23,6 +26,7 @@
 from .api.download_station import SynoDownloadStation
 from .api.dsm.information import SynoDSMInformation
 from .api.dsm.network import SynoDSMNetwork
+from .api.file_station import SynoFileStation
 from .api.photos import SynoPhotos
 from .api.storage.storage import SynoStorage
 from .api.surveillance_station import SynoSurveillanceStation
@@ -95,6 +99,7 @@ def __init__(
         }
         self._download: SynoDownloadStation | None = None
         self._external_usb: SynoCoreExternalUSB | None = None
+        self._file: SynoFileStation | None = None
         self._information: SynoDSMInformation | None = None
         self._network: SynoDSMNetwork | None = None
         self._photos: SynoPhotos | None = None
@@ -246,6 +251,26 @@ async def post(
         """Handles API POST request."""
         return await self._request("POST", api, method, params, **kwargs)
 
+    async def post_upload(
+        self,
+        api: str,
+        method: str,
+        filepath: str,
+        filename: str,
+        content: bytes | BufferedReader | AsyncIterator[bytes],
+        **kwargs: Any,
+    ) -> bytes | dict | str:
+        """Handles an upload API POST request."""
+        return await self._request(
+            "POST",
+            api,
+            method,
+            filepath=filepath,
+            filename=filename,
+            content=content,
+            **kwargs,
+        )
+
     async def generate_url(
         self,
         api: str,
@@ -313,6 +338,7 @@ async def _request(
         url, params, kwargs = await self._prepare_request(api, method, params, **kwargs)
 
         # Request data
+        self._debuglog("---------------------------------------------------------")
         self._debuglog("API: " + api)
         self._debuglog("Request Method: " + request_method)
         response = await self._execute_request(request_method, url, params, **kwargs)
@@ -352,6 +378,34 @@ async def _execute_request(
                 response = await self._session.get(
                     url_encoded, timeout=self._aiohttp_timeout, **kwargs
                 )
+            elif (
+                method == "POST"
+                and (content := kwargs.get("content"))
+                and (filepath := kwargs.get("filepath"))
+                and (filename := kwargs.get("filename"))
+            ):
+                if TYPE_CHECKING:
+                    assert isinstance(content, bytes)  # noqa: S101
+                    assert isinstance(filename, str)  # noqa: S101
+
+                boundary = md5(
+                    str(url_encoded).encode("utf-8"), usedforsecurity=False
+                ).hexdigest()
+                with MultipartWriter("form-data", boundary=boundary) as mp:
+                    part = mp.append(filepath)
+                    part.headers.pop(hdrs.CONTENT_TYPE)
+                    part.set_content_disposition("form-data", name="path")
+
+                    part = mp.append(content)
+                    part.headers.pop(hdrs.CONTENT_TYPE)
+                    part.set_content_disposition(
+                        "form-data", name="file", filename=filename
+                    )
+                    part.headers.add(hdrs.CONTENT_TYPE, "application/octet-stream")
+
+                    response = await self._session.post(
+                        url_encoded, timeout=self._aiohttp_timeout, data=mp
+                    )
             elif method == "POST":
                 data = {}
                 if params is not None:
@@ -366,12 +420,13 @@ async def _execute_request(
                 )
 
             # mask sesitive parameters
-            if _LOGGER.isEnabledFor(logging.DEBUG):
+            if _LOGGER.isEnabledFor(logging.DEBUG) or self._debugmode:
                 response_url = response.url  # pylint: disable=E0606
                 for param in SENSITIV_PARAMS:
                     if params is not None and params.get(param):
                         response_url = response_url.update_query({param: "*********"})
                 self._debuglog("Request url: " + str(response_url))
+                self._debuglog("Request headers: " + str(response.request_info.headers))
                 self._debuglog("Response status_code: " + str(response.status))
                 self._debuglog("Response headers: " + str(dict(response.headers)))
 
@@ -455,6 +510,9 @@ def reset(self, api: SynoBaseApi | str) -> bool:
             if api == SynoCoreExternalUSB.API_KEY:
                 self._external_usb = None
                 return True
+            if api == SynoFileStation.API_KEY:
+                self._file = None
+                return True
             if api == SynoCoreSecurity.API_KEY:
                 self._security = None
                 return True
@@ -488,6 +546,9 @@ def reset(self, api: SynoBaseApi | str) -> bool:
         if isinstance(api, SynoCoreExternalUSB):
             self._external_usb = None
             return True
+        if isinstance(api, SynoFileStation):
+            self._file = None
+            return True
         if isinstance(api, SynoCoreSecurity):
             self._security = None
             return True
@@ -534,6 +595,13 @@ def external_usb(self) -> SynoCoreExternalUSB:
             self._external_usb = SynoCoreExternalUSB(self)
         return self._external_usb
 
+    @property
+    def file(self) -> SynoFileStation:
+        """Gets NAS files."""
+        if not self._file:
+            self._file = SynoFileStation(self)
+        return self._file
+
     @property
     def information(self) -> SynoDSMInformation:
         """Gets NAS informations."""