1+ import getpass
2+ from typing import Tuple , Union
3+ from urllib .parse import urljoin
4+
5+ from requests .models import Response
6+
7+ from pythonanywhere_core .base import call_api , get_api_endpoint
8+ from pythonanywhere_core .exceptions import PythonAnywhereApiException
9+
10+
11+ class Files :
12+ """ Interface for PythonAnywhere files API.
13+
14+ Uses `pythonanywhere_core.base` :method: `get_api_endpoint` to
15+ create url, which is stored in a class variable `Files.base_url`,
16+ then calls `call_api` with appropriate arguments to execute files
17+ action.
18+
19+ Covers:
20+ - GET, POST and DELETE for files path endpoint
21+ - POST, GET and DELETE for files sharing endpoint
22+ - GET for tree endpoint
23+
24+ "path" methods:
25+ - use :method: `Files.path_get` to get contents of file or
26+ directory from `path`
27+ - use :method: `Files.path_post` to upload or update file at given
28+ `dest_path` using contents from `source`
29+ - use :method: `Files.path_delete` to delete file/directory on on
30+ given `path`
31+
32+ "sharing" methods:
33+ - use :method: `Files.sharing_post` to enable sharing a file from
34+ `path` (if not shared before) and get a link to it
35+ - use :method: `Files.sharing_get` to get sharing url for `path`
36+ - use :method: `Files.sharing_delete` to disable sharing for
37+ `path`
38+
39+ "tree" method:
40+ - use :method: `Files.tree_get` to get list of regular files and
41+ subdirectories of a directory at `path` (limited to 1000 results)
42+ """
43+
44+ base_url = get_api_endpoint ().format (username = getpass .getuser (), flavor = "files" )
45+ path_endpoint = urljoin (base_url , "path" )
46+ sharing_endpoint = urljoin (base_url , "sharing/" )
47+ tree_endpoint = urljoin (base_url , "tree/" )
48+
49+ def _error_msg (self , result : Response ) -> str :
50+ """TODO: error responses should be unified at the API side """
51+
52+ if "application/json" in result .headers .get ("content-type" , "" ):
53+ jsn = result .json ()
54+ msg = jsn .get ("detail" ) or jsn .get ("message" ) or jsn .get ("error" , "" )
55+ return f": { msg }
56+ return ""
57+
58+ def path_get (self , path : str ) -> Union [dict , bytes ]:
59+ """Returns dictionary of directory contents when `path` is an
60+ absolute path to of an existing directory or file contents if
61+ `path` is an absolute path to an existing file -- both
62+ available to the PythonAnywhere user. Raises when `path` is
63+ invalid or unavailable."""
64+
65+ url = f"{ self .path_endpoint } { path }
66+
67+ result = call_api (url , "GET" )
68+
69+ if result .status_code == 200 :
70+ if "application/json" in result .headers .get ("content-type" , "" ):
71+ return result .json ()
72+ return result .content
73+
74+ raise PythonAnywhereApiException (
75+ f"GET to fetch contents of { url } { result } { self ._error_msg (result )}
76+ )
77+
78+ def path_post (self , dest_path : str , content : bytes ) -> int :
79+ """Uploads contents of `content` to `dest_path` which should be
80+ a valid absolute path of a file available to a PythonAnywhere
81+ user. If `dest_path` contains directories which don't exist
82+ yet, they will be created.
83+
84+ Returns 200 if existing file on PythonAnywhere has been
85+ updated with `source` contents, or 201 if file from
86+ `dest_path` has been created with those contents."""
87+
88+ url = f"{ self .path_endpoint } { dest_path }
89+
90+ result = call_api (url , "POST" , files = {"content" : content })
91+
92+ if result .ok :
93+ return result .status_code
94+
95+ raise PythonAnywhereApiException (
96+ f"POST to upload contents to { url } { result } { self ._error_msg (result )}
97+ )
98+
99+ def path_delete (self , path : str ) -> int :
100+ """Deletes the file at specified `path` (if file is a
101+ directory it will be deleted as well).
102+
103+ Returns 204 on sucess, raises otherwise."""
104+
105+ url = f"{ self .path_endpoint } { path }
106+
107+ result = call_api (url , "DELETE" )
108+
109+ if result .status_code == 204 :
110+ return result .status_code
111+
112+ raise PythonAnywhereApiException (
113+ f"DELETE on { url } { result } { self ._error_msg (result )}
114+ )
115+
116+ def sharing_post (self , path : str ) -> Tuple [int , str ]:
117+ """Starts sharing a file at `path`.
118+
119+ Returns a tuple with a status code and sharing link on
120+ success, raises otherwise. Status code is 201 on success, 200
121+ if file has been already shared."""
122+
123+ url = self .sharing_endpoint
124+
125+ result = call_api (url , "POST" , json = {"path" : path })
126+
127+ if result .ok :
128+ return result .status_code , result .json ()["url" ]
129+
130+ raise PythonAnywhereApiException (
131+ f"POST to { url } { path } { result } { self ._error_msg (result )}
132+ )
133+
134+ def sharing_get (self , path : str ) -> str :
135+ """Checks sharing status for a `path`.
136+
137+ Returns url with sharing link if file is shared or an empty
138+ string otherwise."""
139+
140+ url = f"{ self .sharing_endpoint } { path }
141+
142+ result = call_api (url , "GET" )
143+
144+ return result .json ()["url" ] if result .ok else ""
145+
146+ def sharing_delete (self , path : str ) -> int :
147+ """Stops sharing file at `path`.
148+
149+ Returns 204 on successful unshare."""
150+
151+ url = f"{ self .sharing_endpoint } { path }
152+
153+ result = call_api (url , "DELETE" )
154+
155+ return result .status_code
156+
157+ def tree_get (self , path : str ) -> dict :
158+ """Returns list of absolute paths of regular files and
159+ subdirectories of a directory at `path`. Result is limited to
160+ 1000 items.
161+
162+ Raises if `path` does not point to an existing directory."""
163+
164+ url = f"{ self .tree_endpoint } { path }
165+
166+ result = call_api (url , "GET" )
167+
168+ if result .ok :
169+ return result .json ()
170+
171+ raise PythonAnywhereApiException (f"GET to { url } { result } { self ._error_msg (result )} )
0 commit comments