|
1 # -*- coding: utf-8 -*- |
|
2 |
|
3 # Copyright (c) 2024 Detlev Offenbach <detlev@die-offenbachs.de> |
|
4 # |
|
5 |
|
6 """ |
|
7 Module implementing the file system request handler of the eric-ide server. |
|
8 """ |
|
9 |
|
10 import os |
|
11 import stat |
|
12 import time |
|
13 |
|
14 from .EricRequestCategory import EricRequestCategory |
|
15 |
|
16 class EricServerFileSystemRequestHandler: |
|
17 """ |
|
18 Class implementing the file system request handler of the eric-ide server. |
|
19 """ |
|
20 |
|
21 def __init__(self, server): |
|
22 """ |
|
23 Constructor |
|
24 |
|
25 @param server reference to the eric-ide server object |
|
26 @type EricServer |
|
27 """ |
|
28 self.__server = server |
|
29 |
|
30 self.__requestMethodMapping = { |
|
31 "Chdir": self.__chdir, |
|
32 "Getcwd": self.__getcwd, |
|
33 "Listdir": self.__listdir, |
|
34 "Mkdir": self.__mkdir, |
|
35 "Rmdir": self.__rmdir, |
|
36 "Replace": self.__replace, |
|
37 "Remove": self.__remove, |
|
38 } |
|
39 |
|
40 def handleRequest(self, request, params, reqestUuid): |
|
41 """ |
|
42 Public method handling the received file system requests. |
|
43 |
|
44 @param request request name |
|
45 @type str |
|
46 @param params dictionary containing the request parameters |
|
47 @type dict |
|
48 @param reqestUuid UUID of the associated request as sent by the eric IDE |
|
49 @type str |
|
50 """ |
|
51 try: |
|
52 result = self.__requestMethodMapping[request](params) |
|
53 self.__server.sendJson( |
|
54 category=EricRequestCategory.FileSystem, |
|
55 reply=request, |
|
56 params=result, |
|
57 reqestUuid=reqestUuid, |
|
58 ) |
|
59 |
|
60 except KeyError: |
|
61 self.__server.sendJson( |
|
62 category=EricRequestCategory.FileSystem, |
|
63 reply=request, |
|
64 params={"Error": f"Request type '{request}' is not supported."}, |
|
65 ) |
|
66 |
|
67 def __chdir(self, params): |
|
68 """ |
|
69 Private method to change the current working directory. |
|
70 |
|
71 @param params dictionary containing the request data |
|
72 @type dict |
|
73 @return dictionary containing the reply data |
|
74 @rtype dict |
|
75 """ |
|
76 directory = params["directory"] |
|
77 try: |
|
78 os.chdir(directory) |
|
79 return {"ok": True} |
|
80 except OSError as err: |
|
81 return { |
|
82 "ok": False, |
|
83 "error": str(err), |
|
84 } |
|
85 |
|
86 def __getcwd(self, params): |
|
87 """ |
|
88 Private method to report the current working directory. |
|
89 |
|
90 @param params dictionary containing the request data |
|
91 @type dict |
|
92 @return dictionary containing the reply data |
|
93 @rtype dict |
|
94 """ |
|
95 return {"directory": os.getcwd()} |
|
96 |
|
97 def __listdir(self, params): |
|
98 """ |
|
99 Private method to report a directory listing. |
|
100 |
|
101 @param params dictionary containing the request data |
|
102 @type dict |
|
103 @return dictionary containing the reply data |
|
104 @rtype dict |
|
105 """ |
|
106 directory = params["directory"] |
|
107 if not directory: |
|
108 directory = os.getcwd() |
|
109 |
|
110 listing = [] |
|
111 for dirEntry in os.scandir(directory): |
|
112 filestat = dirEntry.stat() |
|
113 entry = { |
|
114 "name": dirEntry.name, |
|
115 "path": dirEntry.path, |
|
116 "is_dir": dirEntry.is_dir(), |
|
117 "is_file": dirEntry.is_file(), |
|
118 "is_link": dirEntry.is_symlink(), |
|
119 "mode": filestat.st_mode, |
|
120 "mode_str": stat.filemode(filestat.st_mode), |
|
121 "size": filestat.st_size, |
|
122 "mtime": time.strftime( |
|
123 "%Y-%m-%d %H:%M:%S", time.localtime(filestat.st_mtime) |
|
124 ), |
|
125 } |
|
126 listing.append(entry) |
|
127 |
|
128 return { |
|
129 "directory": directory, |
|
130 "listing": listing, |
|
131 "separator": os.sep, |
|
132 } |
|
133 |
|
134 def __mkdir(self, params): |
|
135 """ |
|
136 Private method to create a new directory. |
|
137 |
|
138 @param params dictionary containing the request data |
|
139 @type dict |
|
140 @return dictionary containing the reply data |
|
141 @rtype dict |
|
142 """ |
|
143 directory = params["directory"] |
|
144 |
|
145 try: |
|
146 os.makedirs(directory) |
|
147 return {"ok": True} |
|
148 except OSError as err: |
|
149 return { |
|
150 "ok": False, |
|
151 "error": str(err), |
|
152 } |
|
153 |
|
154 def __rmdir(self, params): |
|
155 """ |
|
156 Private method to delete a directory. |
|
157 |
|
158 @param params dictionary containing the request data |
|
159 @type dict |
|
160 @return dictionary containing the reply data |
|
161 @rtype dict |
|
162 """ |
|
163 directory = params["directory"] |
|
164 |
|
165 try: |
|
166 os.rmdir(directory) |
|
167 return {"ok": True} |
|
168 except OSError as err: |
|
169 return { |
|
170 "ok": False, |
|
171 "error": str(err), |
|
172 } |
|
173 |
|
174 def __replace(self, params): |
|
175 """ |
|
176 Private method to replace (rename) a file or directory. |
|
177 |
|
178 |
|
179 @param params dictionary containing the request data |
|
180 @type dict |
|
181 @return dictionary containing the reply data |
|
182 @rtype dict |
|
183 """ |
|
184 oldName = params["old_name"] |
|
185 newName = params["new_name"] |
|
186 |
|
187 try: |
|
188 os.replace(oldName, newName) |
|
189 return {"ok": True} |
|
190 except OSError as err: |
|
191 return { |
|
192 "ok": False, |
|
193 "error": str(err), |
|
194 } |
|
195 |
|
196 def __remove(self, params): |
|
197 """ |
|
198 Private method to delete a directory. |
|
199 |
|
200 @param params dictionary containing the request data |
|
201 @type dict |
|
202 @return dictionary containing the reply data |
|
203 @rtype dict |
|
204 """ |
|
205 filename = params["filename"] |
|
206 |
|
207 try: |
|
208 os.remove(filename) |
|
209 return {"ok": True} |
|
210 except OSError as err: |
|
211 return { |
|
212 "ok": False, |
|
213 "error": str(err), |
|
214 } |