Version updates and remove obsolete file methods

.github/workflows/pcloud-test.yml

@@ -14,22 +14,22 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: [3.9, "3.10", "3.11", "3.12", "3.13"]
+        python-version: ["3.10", "3.11", "3.12", "3.13", "3.14"]
       max-parallel: 1
     steps:
     - uses: actions/checkout@v4
       with:
         fetch-depth: 0
     - name: Setup firefox
-      uses: browser-actions/setup-firefox@latest
+      uses: browser-actions/setup-firefox@5914774dda97099441f02628f8d46411fcfbd208
     - name: Set up Python ${{ matrix.python-version }}
-      uses: actions/setup-python@v4
+      uses: actions/setup-python@e797f83bcb11b83ae66e0230d6156d7c80228e7c
       with:
         python-version: ${{ matrix.python-version }}
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip
-        pip install pytest-rerunfailures==15.0 pytest-cov==6.0.0 pytest-timeout==2.3.1
+        pip install pytest-rerunfailures==15.0 pytest-cov==7.0.0 pytest-timeout==2.4.0
         pip install -e ".[test]"
         playwright install
     - name: Lint with flake8
@@ -47,7 +47,7 @@ jobs:
         PCLOUD_OAUTH2_CLIENT_ID: ${{ secrets.PCLOUD_OAUTH2_CLIENT_ID }}
         PCLOUD_OAUTH2_CLIENT_SECRET: ${{ secrets.PCLOUD_OAUTH2_CLIENT_SECRET }}
     - name: Analyze with SonarCloud
-      uses: SonarSource/sonarcloud-github-action@master
+      uses: SonarSource/sonarqube-scan-action@689fb39b34b9aa95ebc5f8f119343ddd51542402
       env:
         GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}  # Needed to get PR information
         SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}   # Generate a token on Sonarcloud.io, add it to the secrets of this repo with the name SONAR_TOKEN (Settings > Secrets > Actions > add new repository secret)

.gitignore

@@ -41,4 +41,6 @@ Pipfile*
 *.whl
 .vscode
 coverage.xml
-.env
+.env
+.ipynb_checkpoints
+etc/

CHANGES.rst

@@ -1,16 +1,22 @@
 Changelog
 =========
 
-
 2.0 (unreleased)
 ----------------
 
+**Breaking changes**
+
+- Remove file API methods,removed from pyCloud API too (see #104) [tomgross]
+- Remove fs support as it does not seem to be maintained any more and it heavily
+  was based on file API methods
+
+**Changes**
+
 - Use `pyproject.toml` instead of `setup.py` [tomgross]
-- Document `eapi`-endpoint for fs.opener [tomgross]
 - Implement undocumented `search`-method (#93) [tomgross]
 - Implement binary-protocol [tomgross]
 - Switch to httpx [tomgross]
-
+- Python compatibility 3.10-3.14 [tomgross]
 
 1.4 (2024-12-29)
 ----------------

README.rst

@@ -10,8 +10,7 @@ This Python **(Version >= 3.6 only!)** library provides a Python API to the pClo
 Features
 ========
 
-- Can be used as a library
-- Provides a PyFileSystem implementation
+- Can be used as a library for accessing pCloud Features exposed by the officiall pCloud API
 
 Examples
 ========
@@ -98,6 +97,16 @@ b) from data:
   >>> img.save(bio, format='jpeg')
   >>> pc.uploadfile(data=bio.getvalue(), filename="image.jpg", path='/path-to-pcloud-dir')
 
+Downloading files
+-----------------
+
+Since the removal of the file API from pCloud downloading file content only via the
+`getzip`- method. There was a convenience method added here to download files:
+
+  >>> pc.file_download(fileid=1234567890)
+  b'xxxx .... zzzz'
+
+
 Searching files
 ---------------
 
@@ -106,40 +115,13 @@ pCloud documentation.
 
   >>> pcapi.search(query="foo", offset=20, limit=10)
 
-PyFilesystem integration
-++++++++++++++++++++++++
-
-Usage of PyFilesystem with opener
-
-  >>> from fs import opener
-  >>> opener.open_fs('pcloud://email%40example.com:SecretPassword@/')
-  <pCloudFS>
-
-Opener of eapi endpoint
-
-  >>> from fs import opener
-  >>> opener.open_fs('pcloud+eapi://email%40example.com:SecretPassword@/')
-  <pCloudFS>
-
-Copying files from Linux to pCloud using PyFilesystem
-
-  >>> from fs import opener, copy
-  >>> with opener.open_fs('pcloud://email%40example.com:SecretPassword@/') as pcloud_fs:
-  >>>    with opener.open_fs('/opt/data_to_copy') as linux_fs:
-  >>>        copy.copy_file(src_fs=linux_fs,
-  >>>                       src_path='database.sqlite3',
-  >>>                       dst_fs=pcloud_fs,
-  >>>                       dst_path='/backup/server/database.sqlite3')
-
-Copy directory from Linux to pCloud using PyFilesystem
+Known issues
+============
 
-  >>> from fs import opener, copy
-  >>> with opener.open_fs('pcloud://email%40example.com:SecretPassword@/') as pcloud_fs:
-  >>>    with opener.open_fs('/opt/data_to_copy') as linux_fs:
-  >>>        copy.copy_dir(src_fs=linux_fs,
-  >>>                      src_path='database/',
-  >>>                      dst_fs=pcloud_fs,
-  >>>                      dst_path='/backup/database/')
+- Some methods (like `listtokens`, `getzip`) are not usable when authenticated via OAuth2
+  (see https://github.com/tomgross/pcloud/issues/61)
+- Since the removal of file system operations from the pCloud API downloading files is 
+  no longer supported when authenticated via OAuth2
 
 Further Documentation
 =====================
@@ -152,15 +134,6 @@ Installation
 
  $ pip install pcloud
 
-Installation with PyFilesystem support
-
- $ bin/pip install pcloud[pyfs]
-
-on zsh (Mac):
-
- $ bin/pip install "pcloud[pyfs]"
-
-
 Development
 ===========
 

pyproject.toml

@@ -17,11 +17,11 @@ classifiers = [
     "Environment :: Other Environment",
     "Environment :: Web Environment",
     "Programming Language :: Python",
-    "Programming Language :: Python :: 3.9",
     "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",
     "Programming Language :: Python :: 3.13",
+    "Programming Language :: Python :: 3.14",
     "Operating System :: OS Independent",
     "License :: OSI Approved :: MIT License",
     "Topic :: Software Development :: Libraries :: Python Modules",
@@ -31,26 +31,18 @@ keywords = ["Python", "pCloud", "REST"]
 
 dependencies = [
     "httpx>=0.28.1",
-    "setuptools>=75.8.0",
+    "setuptools>=80.9.0",
     "wheel>=0.45.1",
 ]
 
 [project.optional-dependencies]
 # development dependency groups
 test = [
-    "pytest==8.3.4",
-    "pytest-sugar==1.0.0",
+    "pytest==8.4.2",
+    "pytest-sugar==1.1.1",
     "wheel==0.45.1",
-    "flake8==7.1.1",
-    "fs==2.4.16",
-    "playwright==1.49.1",
-    "multipart==1.2.1",
-    "zipp>=3.19.1"
+    "flake8==7.3.0",
+    "playwright==1.55.0",
+    "multipart==1.3.0",
+    "zipp>=3.23.0"
 ]
-
-pyfs = ['fs']
-
-[project.entry-points."fs.opener"]
-pcloud = "pcloud.pcloudfs:PCloudOpener"
-"pcloud+eapi"  = "pcloud.pcloudfs:PCloudOpener"
-

src/pcloud/api.py

@@ -1,3 +1,4 @@
+import io
 import os
 import httpx
 import zipfile
@@ -22,13 +23,6 @@
 from urllib.parse import urlunsplit
 
 
-# File open flags https://docs.pcloud.com/methods/fileops/file_open.html
-O_WRITE = int("0x0002", 16)
-O_CREAT = int("0x0040", 16)
-O_EXCL = int("0x0080", 16)
-O_TRUNC = int("0x0200", 16)
-O_APPEND = int("0x0400", 16)
-
 ONLY_PCLOUD_MSG = "This method can't be used from web applications. Referrer is restricted to pcloud.com."
 
 
@@ -60,9 +54,7 @@ def __init__(
     ):
         if endpoint not in self.endpoints:
             log.error(
-                "Endpoint (%s) not found. Use one of: %s",
-                endpoint,
-                ", ".join(self.endpoints.keys()),
+                f"Endpoint ({endpoint}) not found. Use one of: {', '.join(self.endpoints.keys())}"
             )
             return
         elif endpoint == "nearest":
@@ -153,12 +145,12 @@ def supportedlanguages(self, **kwargs):
         return self._do_request("supportedlanguages")
 
     def getnearestendpoint(self):
-        default_api = self.endpoints.get("api")
+        default_api = self.endpoints.get("api").endpoint
         resp = self._do_request(
             "getapiserver", authenticate=False, endpoint=default_api
         )
 
-        api = resp.get("api")
+        api = resp.get("api","")
         if len(api):
             return urlunsplit(["https", api[0], "/", "", ""])
         else:
@@ -331,59 +323,6 @@ def gethlslink(self, **kwargs):
     def gettextfile(self, **kwargs):
         raise OnlyPcloudError(ONLY_PCLOUD_MSG)
 
-    # File API methods
-    @RequiredParameterCheck(("flags",))
-    def file_open(self, **kwargs):
-        return self._do_request("file_open", use_session=True, **kwargs)
-
-    @RequiredParameterCheck(("fd", "count"))
-    def file_read(self, **kwargs):
-        return self._do_request("file_read", json=False, use_session=True, **kwargs)
-
-    @RequiredParameterCheck(("fd",))
-    def file_pread(self, **kwargs):
-        return self._do_request("file_pread", json=False, use_session=True, **kwargs)
-
-    @RequiredParameterCheck(("fd", "data"))
-    def file_pread_ifmod(self, **kwargs):
-        return self._do_request(
-            "file_pread_ifmod", json=False, use_session=True, **kwargs
-        )
-
-    @RequiredParameterCheck(("fd",))
-    def file_size(self, **kwargs):
-        return self._do_request("file_size", use_session=True, **kwargs)
-
-    @RequiredParameterCheck(("fd",))
-    def file_truncate(self, **kwargs):
-        return self._do_request("file_truncate", use_session=True, **kwargs)
-
-    @RequiredParameterCheck(("fd", "data"))
-    def file_write(self, **kwargs):
-        files = [("file", ("upload-file.io", BytesIO(kwargs.pop("data"))))]
-        kwargs["fd"] = str(kwargs["fd"])
-        return self.connection.upload("file_write", files, **kwargs)
-
-    @RequiredParameterCheck(("fd",))
-    def file_pwrite(self, **kwargs):
-        return self._do_request("file_pwrite", use_session=True, **kwargs)
-
-    @RequiredParameterCheck(("fd",))
-    def file_checksum(self, **kwargs):
-        return self._do_request("file_checksum", use_session=True, **kwargs)
-
-    @RequiredParameterCheck(("fd",))
-    def file_seek(self, **kwargs):
-        return self._do_request("file_seek", use_session=True, **kwargs)
-
-    @RequiredParameterCheck(("fd",))
-    def file_close(self, **kwargs):
-        return self._do_request("file_close", use_session=True, **kwargs)
-
-    @RequiredParameterCheck(("fd",))
-    def file_lock(self, **kwargs):
-        return self._do_request("file_lock", use_session=True, **kwargs)
-
     # Archiving
     @RequiredParameterCheck(("path", "fileid"))
     @RequiredParameterCheck(("topath", "tofolderid"))
@@ -493,12 +432,13 @@ def trash_list(self, **kwargs):
     def trash_clear(self, **kwargs):
         return self._do_request("trash_clear", **kwargs)
 
+    @RequiredParameterCheck(("fileid", "folderid"))
     def trash_restorepath(self, **kwargs):
-        raise NotImplementedError
+        return self._do_request("trash_restorepath", **kwargs)
 
     @RequiredParameterCheck(("fileid", "folderid"))
     def trash_restore(self, **kwargs):
-        raise NotImplementedError
+        return self._do_request("trash_restore", **kwargs)
 
     # convenience methods
     @RequiredParameterCheck(("query",))
@@ -508,15 +448,33 @@ def search(self, **kwargs):
     @RequiredParameterCheck(("path",))
     def file_exists(self, **kwargs):
         path = kwargs["path"]
-        resp = self.file_open(path=path, flags=O_APPEND)
+        resp = self.stat(path=path)
         result = resp.get("result")
         if result == 0:
-            self.file_close(fd=resp["fd"])
             return True
-        elif result == 2009:
+        elif result in (2001, 2055):
             return False
         else:
             raise OSError(f"pCloud error occured ({result}) - {resp['error']}:  {path}")
 
+    @RequiredParameterCheck(("fileid",))
+    def file_download(self, **kwargs):
+        fileid = kwargs.get("fileid")
+        resp = self.stat(fileid=fileid, use_session=True)
+        result = resp.get("result")
+        if result == 0:
+            filename = resp["metadata"]["name"]
+        else:
+            raise OSError(
+                f"pCloud error occured ({result}) - {resp.get('error','')}:  {fileid}"
+            )
+
+        zip_bytes = self.getzip(fileids=[fileid], use_session=True)
+        try:
+            with zipfile.ZipFile(io.BytesIO(zip_bytes)) as zf:
+                return zf.read(filename)
+        except zipfile.BadZipFile:
+            raise OSError(f"Data: {zip_bytes}")
+
 
 # EOF

src/pcloud/dummyprotocol.py

@@ -12,6 +12,9 @@ def get(self, url, **kwargs):
     def json(self):
         return self.kwargs
 
+    def raise_for_status(self):
+        """ method to comply with Session API """
+        pass
 
 class PCloudDummyConnection(PCloudJSONConnection):
     """Connection to pcloud.com based on their JSON protocol."""