PyXMake Developer Guide 1.0
PyXMake
Loading...
Searching...
No Matches
__init__.py
1# -*- coding: utf-8 -*-
2# %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
3# % API Module - Classes and Functions %
4# %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
5"""
6Contains all classes and functions to define a web interface
7
8@note: PyXMake module
9Created on 19.11.2020
10
11@version: 1.0
12----------------------------------------------------------------------------------------------
13@requires:
14 -
15
16@change:
17 -
18
19@author: garb_ma [DLR-FA,STM Braunschweig]
20----------------------------------------------------------------------------------------------
21"""
22
23## @package PyXMake.API
24# Contains all classes and functions to create a web application.
25## @author
26# Marc Garbade
27## @date
28# 19.11.2020
29## @par Notes/Changes
30# - Added documentation // mg 19.11.2020
31
32import os, sys
33import abc, six
34import platform
35import shutil
36import socket
37import time
38import io
39import ntpath
40import datetime
41import posixpath
42import zipfile
43import random
44import uuid
45import urllib
46
47try:
48 # API requires 3.6 and higher
49 import uvicorn
50 import json
51 import base64
52 import requests
53
54 from fastapi import FastAPI, APIRouter, File, UploadFile, Query, Body
55 from starlette.responses import Response, RedirectResponse, FileResponse, JSONResponse, HTMLResponse
56 from fastapi.encoders import jsonable_encoder #@UnresolvedImport
57 from starlette.staticfiles import StaticFiles
58 from starlette.requests import Request
59 from enum import Enum
60 from starlette.middleware.cors import CORSMiddleware
61 from starlette.exceptions import HTTPException as StarletteHTTPException
62 from typing import List, Optional, Any, Dict #@UnresolvedImport
63 from pydantic import SecretStr #@UnresolvedImport
64
65except ImportError: pass
66
67try:
68 import PyXMake
69except ImportError:
70 # Script is executed as a plug-in
71 sys.path.insert(0,os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
72 import PyXMake
73finally:
74 from PyXMake.Tools import Utility #@UnresolvedImport
75 from PyXMake import VTL #@UnresolvedImport
76
77## @class PyXMake.API.Base
78# Abstract base class for all API objects. Inherited from built-in ABCMeta & FastAPI.
79# Only compatible with Python 3.6+
80@six.add_metaclass(abc.ABCMeta)
81class Base(object):
82 """
83 Parent class for graphical user interface objects.
84 """
85 def __init__(self, *args, **kwargs):
86 """
87 Low-level initialization of parent class.
88 """
89 # Initialize a new TK main window widget.
90 self.API = FastAPI(*args,**kwargs); self.Router = APIRouter();
91 self.APIObjectKind = "Base"
92
93 def RedirectException(self, url):
94 @self.API.exception_handler(StarletteHTTPException)
95 def custom_http_exception_handler(request, exc):
96 return RedirectResponse(url=url)
97
98 def StaticFiles(self, url, path, index="index.html", html=True):
99 """
100 Serve additional static files. Mount them appropriately.
101 """
102 if os.path.isdir(path) and os.path.exists(os.path.join(path,index)):
103 # Check if directory contains an index.html. If not, create one.
104 if not index.startswith("index") and not os.path.exists(os.path.join(path,"index.html")) and html: # pragma: no cover
105 shutil.copy(os.path.join(path,index),os.path.join(path,"index.html"))
106 # Mount directory with static files to given url, serving index.html if required.
107 self.mount(url, StaticFiles(directory=path, html=html))
108
109 def mount(self, *args):
110 """
111 Return current API's main instance handle.
112 """
113 self.API.mount(*args)
114
115 def include(self, *args):
116 """
117 Return current API's main instance handle.
118 """
119 self.API.include_router(*args);
120
121 def create(self):
122 """
123 Return current API's main instance handle.
124 """
125 # Include all defined routers
126 self.include(self.Router)
127 # Return current API's handle
128 return self.API
129
130 def run(self, Hostname=str(platform.node()), PortID=8020): # pragma: no cover
131 """
132 Run the current API.
133 """
134 # Current API is the main API
135 handle = self.create()
136 # Run current API as main instance
137 uvicorn.run(handle, host=Hostname, port=PortID)
138 pass
139
140## @class PyXMake.API.Backend
141# Class instance to define PyXMake's web API instance
143 """
144 Class instance to define PyXMake's server instance for a web API.
145 """
146 @abc.abstractmethod
147 def __init__(self, *args, **kwargs):
148 """
149 Initialization of PyXMake's Backend API.
150 """
151 # Establish all methods
152 super(Backend, self).__init__()
153 # Collect all hidden attributes
154 __attributes__ = {x:self.APIObjectKind for x in dir(self) if self.APIObjectKind in x}
155 # Set ObjectKind
156 self.APIObjectKind = "Backend"
157 # Definition of all predefined path variables
158 Archive = Enum("ArchiveObjectKind",{"zip":"zip","tar":"tar","gzip":"gzip","lzma":"lzma","bz2":"bz2"})
159 ## Adding support for external file manager.
160 FileManager = Enum("FileObjectKind",{"full":"all","local":"private","shared":"public"})
161 # Merge all class attributes
162 for key, value in __attributes__.items(): setattr(self, key.replace(value,self.APIObjectKind), getattr(self, key))
163 # Delete since not used anymore
164 del __attributes__
165 # %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
166 # % PyXMake - Guide %
167 # %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
168 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","documentation"]), tags=[str(self.__pyx_guide)])
169 def api_self_guide():
170 """
171 Method pointing to the main documentation of the API.
172 """
173 return RedirectResponse(url=self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","documentation",""]))
174
175 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"user","documentation"]), tags=[str(self.__pyx_guide)],response_class=HTMLResponse)
176 def api_user_guide():
177 """
178 Method pointing to the user documentation of the underlying package.
179 """
180 response = str(self.__pyx_url_path)
181 # Build HTML response
182 html_content = \
183 """
184 <!DOCTYPE html><html lang="en"><head><meta charset="UTF-8">
185 <meta http-equiv="refresh" content="1;url="""+'"'+response+'"'+""" />
186 <script>setTimeout(function() {window.location.href = """+'"'+response+'"'+""";}, 1000);
187 </script></head><body></body></html>
188 """
189 return HTMLResponse(html_content)
190
191 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"dev","documentation"]), tags=[str(self.__pyx_guide)])
192 def api_dev_guide():
193 """
194 Method pointing to the developer documentation of the underlying package.
195 """
196 return RedirectResponse(url=self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"dev","documentation",""]))
197
198 # %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
199 # % PyXMake - Interface %
200 # %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
201 @self.Router.patch(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","abaqus","{kind}"]), tags=[str(self.__pyx_interface)])
202 def api_abaqus(kind: Archive, BuildID: str, Source: List[str] = Query(["mcd_astandard"]), ZIP: UploadFile = File(...)):
203 """
204 API for PyXMake to create an ABAQUS compatible Fortran library for Windows machines by using the Intel Fortran Compiler.
205 """
206 # Check if an old file is present in the current workspace. Delete it.
207 if os.path.exists(os.path.join(VTL.Scratch,ZIP.filename)):
208 os.remove(os.path.join(VTL.Scratch,ZIP.filename))
209 # Everything is done within a temporary directory. Update the uploaded ZIP folder with new content. Delete the input.
210 with Utility.TemporaryDirectory(VTL.Scratch), Utility.UpdateZIP(ZIP.filename, ZIP.file, VTL.Scratch, update=False):
211 # Get all relevant uploaded files.
212 files = Utility.FileWalk(Source, path=os.getcwd())
213 # Monitor the build process. Everything is returned to a file
214 with Utility.FileOutput('result.log'):
215 VTL.abaqus(BuildID, files, source=os.getcwd(), scratch=os.getcwd(), output=os.getcwd(), verbosity=2)
216 # Present result to FAST API
217 return FileResponse(path=os.path.join(VTL.Scratch,ZIP.filename),filename=ZIP.filename)
218
219 @self.Router.patch(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","client","{kind}"]), tags=[str(self.__pyx_interface)],
220 description=
221 """
222 API for PyXMake. Connect with a remote OpenAPI generator service to construct a client library from an Open API scheme given as either a file or an accessible URL.
223 The result is returned as a ZIP folder. If no ZIP folder has been provided, one is generated in the process.
224 """ )
225 def api_client(
226 kind: Archive,
227 request: Request,
228 URI: str = Query(...,
229 description=
230 """
231 The name of the main input file present in the supplied ZIP archive or the full URL address. The API scheme will be downloaded automatically.
232 Note that the URI is always interpreted as an URL if a file with the name cannot be found in the archive.
233 """
234 ),
235 ClientID: Optional[str] = Query("python",
236 description=
237 """
238 This is the generated client format. Defaults to python.
239 """
240 ),
241 CLI: List[str] = Query(["--skip-validate-spec"],
242 description=
243 """
244 Additional command line arguments. Please refer to the official documentation of OpenAPI generator to get a glimpse of all available options.
245 """
246 ),
247 ZIP: UploadFile = File(None,
248 description=
249 """
250 An compressed archive containing all additional files required for this job. This archive is updated and returned. All input files are preserved.
251 It additionally acts as an on-demand temporary user directory. It prohibits access from other users preventing accidental interference.
252 However, the scratch workspace is cleaned constantly. So please download your result immediately.
253 """
254 )):
255 """
256 API to connect to the OpenAPI client generator.
257 """
258 import tempfile
259 import subprocess
260
261 from pathlib import Path
262 from packaging.version import parse
263 # Local variables
264 delimn = ".";
265
266 # Create some temporary variables
267 canditate = str(next(tempfile._get_candidate_names()));
268
269 # Check if a ZIP folder has been given. Create a default entry (empty archive)
270 if not ZIP: ZIP = Utility.UpdateZIP.create(delimn.join([canditate,"zip"]))
271
272 try:
273 # Check if an old file is present in the current workspace. Delete it.
274 if os.path.exists(os.path.join(VTL.Scratch,ZIP.filename)): os.remove(os.path.join(VTL.Scratch,ZIP.filename))
275 except: pass
276
277 # Everything is done within a temporary directory. Update the uploaded ZIP folder with new content. Preserve the input.
278 with Utility.TemporaryDirectory(VTL.Scratch), Utility.UpdateZIP(ZIP.filename, ZIP.file, VTL.Scratch, update=True):
279 root = Utility.AsDrive("mnt")
280 p = Path(os.path.abspath(os.getcwd())).parts; path = posixpath.join(root, *p[1 if Utility.GetPlatform() in ["windows"] else 2:])
281
282 # This is the required base image
283 image = "docker.io/openapitools/openapi-generator-cli"
284
285 # Create a common mount point
286 mount = [{"ReadOnly": False,"Source": "stmlab_scratch","Target": "/mnt","Type":"volume"}]
287
288 # Create base command
289 openapi_cli = ["docker","run","--rm","-v","/mnt:%s" % root,image.split(posixpath.sep, 1)[-1]]
290
291 # URI is not a file (or does not exist in the directory). Attempt to download the resource. Fail if unsuccessful
292 if not os.path.exists(URI):
293 response = requests.get(URI)
294 with open(Utility.PathLeaf(URI), "w") as f: f.write(response.text)
295 URI = Utility.PathLeaf(URI)
296 ## Update base URL for client to automatically connect to the server from which is setup is derived. This is only meaningful
297 # when a definition file is published alongside with a public API interface. Does not affect JSON files presented directly through
298 # the ZIP archive
299 try:
300 # If response was successful, URL now contains the servers net location
301 url = '{uri.scheme}://{uri.netloc}'.format(uri= urllib.parse.urlparse(response.url))
302 # Opening OpenAPI definition as JSON file
303 with open(URI,'r') as openfile: openapi_file = json.load(openfile)
304 # Update the default list of servers to include the complete URL by default
305 servers = [{'url': posixpath.sep.join([url,Utility.PathLeaf(x["url"])]).rstrip(posixpath.sep)} for x in openapi_file.get("servers",[{"url":""}])]
306 # Restrict version of specification for now
307 if parse(openapi_file.get("openapi")) >= parse("3.1.0"): openapi_file.update({'openapi':'3.0.3'})
308 # Update configuration
309 openapi_file.update({'servers':servers})
310 # Overwrite existing file
311 with open(URI,'w') as openfile: openfile.write(json.dumps(openapi_file))
312 except: pass
313
314 # If Java is available - use generator directly.
315 if Utility.GetExecutable("java") and Utility.GetOpenAPIGenerator():
316 ## Java executable in a Docker container should take precedence
317 openapi_cli = [Utility.GetExecutable("java", get_path=True, path=os.pathsep.join(['/usr/bin'] if all([
318 Utility.IsDockerContainer(), Utility.GetPlatform() in ["linux"]]) else [] + os.getenv("PATH",os.defpath).split(os.pathsep)))[-1],"-jar",
319 Utility.GetOpenAPIGenerator()];
320
321 # The command line used in the container
322 if not openapi_cli[0] in ["docker"]: command = ["generate","-i",posixpath.join(os.getcwd(),URI),"-g",str(ClientID),"-o",posixpath.join(os.getcwd(),str(ClientID))]
323 else: command = ["generate","-i",posixpath.join(path,URI),"-g",str(ClientID),"-o",posixpath.join(path,str(ClientID))]
324 # Merge executable and command
325 command = openapi_cli + command
326 # Add user-specific command line options
327 ind = command.index("generate") + 1; command[ind:ind] = CLI
328
329 # Assemble request body
330 data = {"image":image,"command":command[6:],"keep":False,"template":{"TaskTemplate":{"ContainerSpec":{"Mounts":mount}}}}
331
332 # Execute Java command
333 if not command[0] in ["docker"]: subprocess.check_call(command)
334 # Lecacy version
335 else:
336 # Attempt direct invocation using remote docker instance
337 requests.post(str(request.url.scheme)+"://%s/0/Lab/UI/Orchestra" % self.APIBase,params={"name":"stm_openapi"}, json=data)
338 # Check if attempt was successful. Attempt orchestra in case of failure.
339 if not os.path.exists(os.path.join(os.getcwd(),ClientID)):
340 r = requests.post(str(request.url.scheme)+"://%s/0/Lab/UI/Orchestra" % self.APIBase,
341 params={"name":"stm_openapi"}, json={"mounts":mount, "command":command})
342 r.raise_for_status()
343
344 # Get all permissions for the current folder
345 if Utility.GetPlatform() not in ["windows"]: subprocess.check_call(["sudo","chmod","-R",str(777),os.path.abspath(os.getcwd())])
346 pass
347 # Present result to FAST API
348 return FileResponse(path=os.path.join(VTL.Scratch,ZIP.filename),filename=ZIP.filename)
349
350 @self.Router.patch(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","cxx","{kind}"]), tags=[str(self.__pyx_interface)])
351 def api_cxx(kind: Archive, string_to_channel_through: str) -> dict:
352 """
353 Dummy function for showing a quick response
354 """
355 # from PyXMake.VTL import cxx
356 string = "hello"
357 return {"return_code": string}
358
359 @self.Router.patch(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","doxygen","{kind}"]), tags=[str(self.__pyx_interface)])
360 def api_doxygen(kind: Archive, string_to_channel_through: str) -> dict:
361 """
362 Dummy function for showing a quick response
363 """
364 # from PyXMake.VTL import doxygen
365 string = "hello"
366 return {"return_code": string}
367
368 @self.Router.patch(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","ifort","{kind}"]), tags=[str(self.__pyx_interface)])
369 def api_ifort(kind: Archive, BuildID: str, Source: List[str] = Query([x for x in VTL.GetSourceCode(0)]), ZIP: UploadFile = File(...)):
370 """
371 API for PyXMake to create a static library for Windows machines by using the Intel Fortran Compiler.
372 """
373 # Check if an old file is present in the current workspace. Delete it.
374 if os.path.exists(os.path.join(VTL.Scratch,ZIP.filename)):
375 os.remove(os.path.join(VTL.Scratch,ZIP.filename))
376 # Everything is done within a temporary directory. Update the uploaded ZIP folder with new content. Delete the input.
377 with Utility.TemporaryDirectory(VTL.Scratch), Utility.UpdateZIP(ZIP.filename, ZIP.file, VTL.Scratch, update=False):
378 # Get all relevant uploaded files.
379 files = Utility.FileWalk(Source, path=os.getcwd())
380 # Create two dummy directories to avoid problems during build event.
381 os.makedirs("include",exist_ok=True); os.makedirs("lib",exist_ok=True)
382 # Monitor the build process. Everything is returned to a file
383 with Utility.FileOutput('result.log'):
384 VTL.ifort(BuildID, files, source=os.getcwd(), scratch=os.getcwd(), make=os.getcwd(), verbosity=2)
385 # Present result to FAST API
386 return FileResponse(path=os.path.join(VTL.Scratch,ZIP.filename),filename=ZIP.filename)
387
388 @self.Router.patch(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","java","{kind}"]), tags=[str(self.__pyx_interface)])
389 def api_java(kind: Archive, BuildID: str, Source: List[str] = Query([x for x in VTL.GetSourceCode(0)]), ZIP: UploadFile = File(...)):
390 """
391 API for PyXMake to create a Java compatible Fortran library for Windows machines by using the Intel Fortran Compiler.
392 """
393 # Check if an old file is present in the current workspace. Delete it.
394 if os.path.exists(os.path.join(VTL.Scratch,ZIP.filename)):
395 os.remove(os.path.join(VTL.Scratch,ZIP.filename))
396 # Everything is done within a temporary directory. Update the uploaded ZIP folder with new content. Delete the input.
397 with Utility.TemporaryDirectory(VTL.Scratch), Utility.UpdateZIP(ZIP.filename, ZIP.file, VTL.Scratch, update=False):
398 # Get all relevant uploaded files.
399 files = Utility.FileWalk(Source, path=os.getcwd())
400 # Monitor the build process. Everything is returned to a file
401 with Utility.FileOutput('result.log'):
402 VTL.java(BuildID, files, source=os.getcwd(), scratch=os.getcwd(), output=os.getcwd(), verbosity=2)
403 # Present result to FAST API
404 return FileResponse(path=os.path.join(VTL.Scratch,ZIP.filename),filename=ZIP.filename)
405
406 @self.Router.patch(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","latex","{kind}"]), tags=[str(self.__pyx_interface)],
407 description=
408 """
409 API for PyXMake. Connect with a remote Overleaf service to compile a given Latex archive. The result PDF is added to the archive and returned.
410 """ )
411 def api_latex(kind: Archive,
412 BuildID: str,
413 ZIP: UploadFile = File(...,
414 description=
415 """
416 An compressed archive containing all additional files required for this job. This archive is updated and returned. All input files are preserved.
417 It additionally acts as an on-demand temporary user directory. It prohibits access from other users preventing accidental interference.
418 However, the scratch workspace is cleaned constantly. So please download your result immediately.
419 """)):
420 """
421 API for PyXMake to compile a Latex document remotely using Overleaf.
422 """
423 # Check if an old file is present in the current workspace. Delete it.
424 if os.path.exists(os.path.join(VTL.Scratch,ZIP.filename)): os.remove(os.path.join(VTL.Scratch,ZIP.filename))
425 # Everything is done within a temporary directory. Update the uploaded ZIP folder with new content. Delete the input.
426 with Utility.TemporaryDirectory(VTL.Scratch):
427 # Collapse temporary spooled file
428 Utility.UpdateZIP(ZIP.filename, ZIP.file, VTL.Scratch, update=False)
429 # Monitor the remote build process. Everything is returned to a file
430 with Utility.FileOutput('result.log'): VTL.latex(BuildID, ZIP.filename, API="Overleaf", verbosity=2, keep=False)
431 # Add results to ZIP archive.
432 archive = zipfile.ZipFile(ZIP.filename,'a'); [archive.write(x, os.path.basename(x)) for x in os.listdir() if not x.endswith(".zip")]; archive.close()
433 shutil.copy(ZIP.filename, VTL.Scratch)
434 # Present result to FAST API
435 return FileResponse(path=os.path.join(VTL.Scratch,ZIP.filename),filename=ZIP.filename)
436
437 @self.Router.patch(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","f2py","{kind}"]), tags=[str(self.__pyx_interface)])
438 def api_py2x(kind: Archive, BuildID: str, Source: List[str] = Query([x for x in VTL.GetSourceCode(0)]), ZIP: UploadFile = File(...)):
439 """
440 API for PyXMake to create a Python compatible Fortran library for Windows machines by using the Intel Fortran Compiler.
441 """
442 # Check if an old file is present in the current workspace. Delete it.
443 if os.path.exists(os.path.join(VTL.Scratch,ZIP.filename)):
444 os.remove(os.path.join(VTL.Scratch,ZIP.filename))
445 # Everything is done within a temporary directory. Update the uploaded ZIP folder with new content. Delete the input.
446 with Utility.TemporaryDirectory(VTL.Scratch), Utility.UpdateZIP(ZIP.filename, ZIP.file, VTL.Scratch, update=False):
447 # Get all relevant uploaded files.
448 files = Utility.FileWalk(Source, path=os.getcwd())
449 # Monitor the build process. Everything is returned to a file
450 with Utility.FileOutput('result.log'):
451 VTL.py2x(BuildID, files, source=os.getcwd(), scratch=os.getcwd(), output=os.getcwd(), verbosity=2)
452 # Present result to FAST API
453 return FileResponse(path=os.path.join(VTL.Scratch,ZIP.filename),filename=ZIP.filename)
454
455 @self.Router.patch(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","ssh_f2py","{kind}"]), tags=[str(self.__pyx_interface)])
456 def api_ssh_f2py(kind: Archive, time_sleep: int) -> dict:
457 """
458 Dummy function for showing a long response time
459 """
460 # from PyXMake.VTL import ssh_f2py
461 return {"return_code":str(time_sleep)}
462
463 @self.Router.patch(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","ssh_ifort","{kind}"]), tags=[str(self.__pyx_interface)])
464 def api_ssh_ifort(kind: Archive, time_sleep: int) -> dict:
465 """
466 Dummy function for showing a long response time
467 """
468 # from PyXMake.VTL import ssh_ifort
469 return {"return_code":str(time_sleep)}
470
471 @self.Router.patch(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","ssh_make","{kind}"]), tags=[str(self.__pyx_interface)])
472 def api_ssh_make(kind: Archive, time_sleep: int) -> dict:
473 """
474 Dummy function for showing a long response time
475 """
476 # from PyXMake.VTL import ssh_make
477 return {"return_code":str(time_sleep)}
478
479 # %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
480 # % PyXMake - Professional %
481 # %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
482 @self.Router.put(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","self"]), tags=[str(self.__pyx_professional)])
483 def api_self_access(Attribute: Optional[str] = "PyXMake.Tools.Utility.IsDockerContainer", args: List[Any] = [], kwargs: Dict[str,Any] = None):
484 """
485 API to remotely access PyXMake methods.
486 """
487 import numpy as np
488 import PyXMake as pyx #@UnresolvedImport
489 from PyXMake.Build.Make import Py2X #@UnresolvedImport
490 # Default status code (Not Documented Error)
491 status_code = 500
492 # Evaluate an arbitrary attribute. Get its callback method.
493 callback = Py2X.callback(pyx,Attribute)
494 with Utility.TemporaryDirectory(): result = [x.tolist() if hasattr(x, "tolist") else x for x in np.atleast_1d(callback(*args,**kwargs))];
495 if result: status_code = 200
496 return JSONResponse(status_code=status_code,content=result,)
497
498 @self.Router.head(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","user","{kind}"]), tags=[str(self.__pyx_professional)],
499 description=
500 """
501 API for user info service. Verifies that a given token can be verified against a given service.
502 """ )
503 def api_get_user(request: Request,
504 kind: Enum("TokenObjectKind",{"portainer":"portainer","shepard":"shepard","gitlab":"gitlab","user":"client"}),
505 Token: SecretStr = Query(...,
506 description=
507 """
508 A valid X-API-Token.
509 """
510 ),
511 ClientID: str = Query(None,
512 description=
513 """
514 An user-defined client. Only meaningful when a non-default client is requested.
515 """
516 )):
517 """
518 API to verify a given token. Returns information about the token holder if exists.
519 """
520 result = {}
521 # Fail gracefully if service is not reachable
522 try:
523 # A non-default client is requested
524 if kind.name in ["user"] and ClientID:
525 r = requests.post("https://token-api.fa-services.intra.dlr.de/decode_and_verify?token=%s&expected_client_id=%s" % (Token.get_secret_value(), ClientID,), verify=False)
526 # A non-default client is requested but not explicitly given
527 elif kind.name in ["user"] and not ClientID: return JSONResponse(status_code=406, content="ClientID cannot be blank in non-default client mode")
528 # A default client is requested
529 else: r = requests.get(str(request.url.scheme)+"://%s/2/PyXMake/api/%s/user" % (self.APIBase,kind.value,),
530 params={"Token":Token.get_secret_value()}, verify=False)
531 result = r.json()
532 except: pass
533 # Modify response with respect to the content of result
534 if result: return JSONResponse(status_code=r.status_code, content=result,)
535 else: return JSONResponse(status_code=404,
536 content="There is no match for the given token and client %s." % str(kind.value.title() if not kind.name in ["user"] else ClientID),)
537
538 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","info","{kind}"]), tags=[str(self.__pyx_professional)],
539 description=
540 """
541 API for user info service. Verifies that a given attribute can be found in a given category.
542 """ )
543 def api_get_info(
544 kind: Enum("UserObjectKind",{"dlr-username":"user","email":"mail"}),
545 attribute: str = Query(...,
546 description="")):
547 """
548 API to verify that a given user actual exists or is another service.
549 """
550 result = []
551 # Fail gracefully if service is not reachable
552 try:
553 from PyCODAC.Tools.IOHandling import Shepard #@UnresolvedImport
554 result = [x for x in requests.get(Shepard.user_url).json() if x[kind.name] == attribute]
555 except: return JSONResponse(status_code=404,content="User info service not available",)
556 # We have got our information
557 status_code = 200;
558 # Found an user. Return the information:
559 if result: return JSONResponse(status_code=status_code,content=result[0],)
560 # Return a message if no user data was found.
561 else: return JSONResponse(status_code=status_code,content="There is no match for the given attribute in the category %s." % kind.value,)
562
563 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","generator","{kind}"]), tags=[str(self.__pyx_professional)], include_in_schema=False,
564 description=
565 """
566 API for generator service. Generates a random response depending on the given path parameter.
567 """ )
568 def api_generator(request: Request, kind: Enum("GeneratorObjectKind",{"message":"message"})):
569 """
570 Generates a random response in dependence of a given path parameter.
571 """
572 import git
573 import ast
574 # Procedure
575 result = ""; gist = "6440b706a97d2dd71574769517e7ed32"
576 # Construct the response
577 try:
578 with Utility.TemporaryDirectory(VTL.Scratch):
579 # We cannot reach the API or reached the limit. Fall back to cloning
580 if not os.path.exists(gist) and not os.getenv("pyx_message",""):
581 git.Repo.clone_from(posixpath.join("https://gist.github.com",gist),gist) #@UndefinedVariable
582 with io.open(os.path.join(gist,"loading_messages.js"),mode="r",encoding="utf-8") as f: result = f.read()
583 # Always refer to the environment variable directly
584 result = result.replace("\n","").replace("export default ","").replace(";","")
585 os.environ["pyx_message"] = result
586 # Remove directory with no further use
587 try: git.rmtree(gist)
588 except: pass
589 # Return only one element
590 result = ast.literal_eval(os.getenv("pyx_message",result)); result = random.choice(result);
591 status_code = 200
592 except:
593 result = "Service not available"; status_code = 500
594 # Return the generated response
595 return JSONResponse(status_code=status_code,content=result,)
596
597 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","database"]), tags=[str(self.__pyx_professional)], include_in_schema=False,
598 description=
599 """
600 API for internal database service. Use to collect information about all available internal database references.
601 """ )
602 def api_database():
603 """
604 API to fetch information about all internal databases.
605 """
606 # Initialize everything
607 result = {}; status_code = 500
608 # Construct the response
609 from PyCODAC.Database import Mongo
610 # Construct a valid JSON response
611 try: result = { "DataContainer" : Mongo.base_table, "DataTypes" : Mongo.base_reference }
612 except: pass
613 # Try to return the response directly.
614 try: return (Enum("DataObjectKind",result["DataContainer"]),result["DataTypes"] )
615 except: pass
616 if result: status_code = 200
617 # Return license servers as a dictionary
618 return JSONResponse(status_code=status_code,content=result,)
619
620 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","license"]), tags=[str(self.__pyx_professional)],
621 description=
622 """
623 API for FlexLM license manager services. Use to collect information about all supported license servers.
624 """ )
625 def api_license(request: Request):
626 """
627 API to fetch FlexLM license server information
628 """
629 # Initialize everything
630 result = {}; status_code = 500
631 # Construct the response
632 try:
633 result["intel"] =[ "28518@INTELSD2.intra.dlr.de" ]
634 result["ansys"] = [ "1055@ansyssz1.intra.dlr.de" ]
635 result["nastran"] = [ "1700@nastransz1.intra.dlr.de", "1700@nastransz2.intra.dlr.de" ]
636 result["abaqus"] = [ "27018@abaqussd1.intra.dlr.de" ]
637 result["hypersizer"] = [ "27010@hypersizersd1.intra.dlr.de" ]
638 result["altair"] = ["47474@altairsz1.intra.dlr.de"]
639 status_code = 200
640 except: pass
641
642 # Return license servers as a dictionary
643 return JSONResponse(status_code=status_code,content=result,)
644
645 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","flexlm"]), tags=[str(self.__pyx_professional)], include_in_schema=False,
646 description=
647 """
648 API for FlexLM license manager services. Returns the complete response as a JSON string.
649 """ )
650 def api_flexlm(license_query: Optional[str] = Query("27018@abaqussd1", description=
651 """
652 Qualified name or IP address and port number of the license server. The license server must be active and supporting FlexLM.
653 """
654 )):
655 """
656 API to fetch FlexLM all license server information
657 """
658 import tempfile
659
660 from collections import defaultdict #@UnresolvedImport
661 from PyXMake.Build import Make #@UnresolvedImport
662 # Default status code (Not Documented Error)
663 status_code = 500
664 # Check whether a connection with FlexLM can be established or not.
665 default_user = "f_testpa"; default_host = "cluster.bs.dlr.de"
666 secrets = os.path.join(Utility.AsDrive("keys"),default_host)
667 try: default_key = os.path.join(secrets,[x for x in os.listdir(secrets) if x.split(".")[-1] == x][0])
668 except: return JSONResponse(status_code=404,content="Service not available",)
669 # Evaluate an arbitrary attribute. Get its callback method.
670 result = defaultdict(list)
671 try:
672 # Operate fully in a temporary directory
673 with Utility.TemporaryDirectory():
674 # Create a temporary file name
675 tmp = str(next(tempfile._get_candidate_names()))
676 command = "/cluster/software/flexlm/bin/lmutil lmstat -c %s -a" % license_query
677 connect = Make.SSH("FlexLM",[]); connect.Settings(default_user, key=default_key, timeout=1)
678 # Execute the SSH script and log the result
679 with Utility.FileOutput(tmp): Utility.SSHPopen(connect.ssh_client, ntpath.pathsep.join([command]),verbosity=2)
680 # Read logging information into memory
681 with open(tmp,"r") as f: content = f.read()
682 # Success
683 result = json.dumps(content); status_code = 200
684 # Prevent endless loops if service is not reachable
685 except TimeoutError: return JSONResponse(status_code=404,content="FlexLM service not reachable",)
686 except: pass
687 # Return license status
688 return JSONResponse(status_code=status_code,content=result,)
689
690 @self.Router.post(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","flexlm","show"]), tags=[str(self.__pyx_professional)],
691 description=
692 """
693 API for FlexLM license manager services. Use to collect license occupancy rate information for a given query and attributes parameters.
694 Defaults to ABAQUS licensing information and occupancy of CAE and solver tokens.
695 """ )
696 def api_flexlm_show(request: Request,
697 license_query: Optional[str] = Query("27018@abaqussd1.intra.dlr.de",
698 description=
699 """
700 Qualified name or IP address and port number of the license server. The license server must be active and supporting FlexLM.
701 """
702 ),
703 license_attributes: List[str] = Body(['abaqus','cae'],
704 description=
705 """
706 List of attributes to be searched for in the FlexLM response. Defaults to ABAQUS CAE and solver token occupancy.
707 """
708 )):
709 """
710 API to fetch FlexLM license occupancy information
711 """
712 from collections import defaultdict #@UnresolvedImport
713
714 # Evaluate an arbitrary attribute. Get its callback method.
715 result = defaultdict(list)
716
717 # Create default request
718 license_request = defaultdict(lambda:[1,0])
719 # Create an dictionary with all requests
720 for x in license_attributes: license_request[x]
721
722 # Fetch content from FlexLM license services
723 r = requests.get(str(request.url.scheme)+"://%s/2/PyXMake/api/flexlm" % self.APIBase,
724 params={"license_query":license_query}, verify=False)
725 if r.status_code == 404: return JSONResponse(status_code=r.status_code,content=r.json(),)
726 elif not r.status_code == 200: return JSONResponse(status_code=r.status_code,content=r.content,)
727 content = json.loads(r.json())
728
729 try:
730 # Default status code (Not Documented Error)
731 status_code = 500
732 # Collect license information
733 for key, value in license_request.items():
734 for output in value: result[key].append(int(content.split("Users of %s:" % key)[1].split("licenses")[output].split("Total of")[1].split()[0]))
735 # Success
736 result = json.dumps(result); status_code = 200
737 except: pass
738 # Return license status
739 return JSONResponse(status_code=status_code,content=result,)
740
741 @self.Router.post(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","flexlm","inspect"]), tags=[str(self.__pyx_professional)],
742 description=
743 """
744 API for FlexLM license manager services. Use to collect information about license holders as well as number of licenses being used by each user.
745 """ )
746 def api_flexlm_inspect(request: Request,
747 license_query: Optional[str] = Query("27018@abaqussd1.intra.dlr.de",
748 description=
749 """
750 Qualified name or IP address and port number of the license server. The license server must be active and supporting FlexLM.
751 """
752 ),
753 license_attributes: List[str] = Body(['abaqus','cae'],
754 description=
755 """
756 List of attributes to be searched for in the FlexLM response. Defaults to ABAQUS CAE and solver token user information.
757 """
758 )):
759 """
760 API to fetch FlexLM license holder information and number of licenses in use.
761 """
762 from collections import defaultdict #@UnresolvedImport
763
764 # Evaluate an arbitrary attribute. Get its callback method.
765 result = defaultdict(list)
766
767 # Fetch content from FlexLM license services
768 r = requests.get(str(request.url.scheme)+"://%s/2/PyXMake/api/flexlm" % self.APIBase,
769 params={"license_query":license_query}, verify=False)
770 if r.status_code == 404: return JSONResponse(status_code=r.status_code,content=r.json(),)
771 elif not r.status_code == 200: return JSONResponse(status_code=r.status_code,content=r.content,)
772 content = json.loads(r.json())
773
774 try:
775 # Default status code (Not Documented Error)
776 status_code = 500
777 # Collect license information
778 for key in license_attributes:
779 result[key].append([x.strip() for x in content.split("Users of %s:" % key)[1].split("Users of")[0].split("\n")[4:] if Utility.IsNotEmpty(x)])
780 # Success
781 result = json.dumps(result); status_code = 200
782 except: pass
783 # Return license holders and detailed job information
784 return JSONResponse(status_code=status_code,content=result,)
785
786 @self.Router.post(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","f2py","inspect"]), tags=[str(self.__pyx_professional)], include_in_schema=False)
787 def api_py2x_inspect(ModuleID: Optional[str] = "mcd_corex64", ZIP: Optional[UploadFile] = File(None)):
788 """
789 Inspect and return all qualified symbols of an existing shared object library.
790 Optionally, upload a compiled Python extension library and return all qualified symbols.
791 """
792 # Local import definitions
793 import importlib, contextlib
794 from PyXMake.Build.Make import Py2X #@UnresolvedImport
795 # Attempt to import PyCODAC. Fail gracefully
796 try: import PyCODAC.Core #@UnresolvedImport
797 except: pass
798 # Check if an old file is present in the current workspace. Delete it.
799 try:
800 if os.path.exists(os.path.join(VTL.Scratch,ZIP.filename)): os.remove(os.path.join(VTL.Scratch,ZIP.filename))
801 filename = ZIP.filename; archive = ZIP.file
802 except:
803 filename ='Default.zip';
804 # Create an empty dummy archive if non has been provided. This is the default.
805 empty_zip_data = b'PK\x05\x06\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
806 with open(os.path.join(VTL.Scratch,filename), 'wb') as ZIP: ZIP.write(empty_zip_data)
807 archive = open(os.path.join(VTL.Scratch,filename),"rb")
808 # Everything is done within a temporary directory. Update the uploaded ZIP folder with new content. Delete the input.
809 with Utility.TemporaryDirectory(VTL.Scratch), Utility.UpdateZIP(filename, archive, VTL.Scratch, update=False):
810 # Search for an supplied shared library. Default to an empty list of non has been found.
811 try: default = str(Utility.PathLeaf([os.path.abspath(x) for x in os.listdir(os.getcwd()) if x.endswith(".pyd")][0]).split(".")[0])
812 except: default = list([])
813 # Import the supplied module.
814 try:
815 if not ModuleID and default: mod = importlib.import_module(default)
816 else: mod = importlib.import_module(ModuleID)
817 except: pass
818 ## Special case. MCODAC is loaded by default. However, when the user supplies a custom version of the library, this should be used instead.
819 # Attempt to reload - but fail gracefully.
820 try:
821 from importlib import reload
822 sys.path.insert(0,os.getcwd())
823 mod = reload(mod)
824 except: pass
825 # Inspect content of the supplied shared library. Return all methods or throw an error.
826 try: AllModules = jsonable_encoder(Py2X.inspect(mod))
827 except: AllModules = JSONResponse(status_code=404,content={"message": "No module named %s" % ModuleID or default},)
828 return AllModules
829
830 @self.Router.post(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","f2py","show"]), tags=[str(self.__pyx_professional)], include_in_schema=False)
831 def api_py2x_show(ModuleID: Optional[str] = "mcd_corex64", Attribute: Optional[str] = ".", ZIP: Optional[UploadFile] = File(None)):
832 """
833 Return the doc string of an existing symbol in a shared library.
834 Optionally, upload a compiled Python extension library.
835 """
836 # Local import definitions
837 import importlib, contextlib
838 from PyXMake.Build.Make import Py2X #@UnresolvedImport
839 # Attempt to import PyCODAC. Fail gracefully
840 try: import PyCODAC.Core #@UnresolvedImport
841 except: pass
842 # Check if an old file is present in the current workspace. Delete it.
843 try:
844 if os.path.exists(os.path.join(VTL.Scratch,ZIP.filename)): os.remove(os.path.join(VTL.Scratch,ZIP.filename))
845 filename = ZIP.filename; archive = ZIP.file
846 except:
847 filename ='Default.zip';
848 # Create an empty dummy archive if non has been provided. This is the default.
849 empty_zip_data = b'PK\x05\x06\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
850 with open(os.path.join(VTL.Scratch,filename), 'wb') as ZIP: ZIP.write(empty_zip_data)
851 archive = open(os.path.join(VTL.Scratch,filename),"rb")
852 # Everything is done within a temporary directory. Update the uploaded ZIP folder with new content. Delete the input.
853 with Utility.TemporaryDirectory(VTL.Scratch), Utility.UpdateZIP(filename, archive, VTL.Scratch, update=False):
854 # Search for an supplied shared library. Default to an empty list of non has been found.
855 try: default = str(Utility.PathLeaf([os.path.abspath(x) for x in os.listdir(os.getcwd())
856 if x.endswith(".pyd" if Utility.GetPlatform() in ["windows"] else ".so")][0]).split(".")[0])
857 except: default = list([])
858 # Import the supplied module.
859 try:
860 if not ModuleID and default: mod = importlib.import_module(default)
861 else: mod = importlib.import_module(ModuleID)
862 except: pass
863 ## Special case. MCODAC is loaded by default. However, when the user supplies a custom version of the library, this should be used instead.
864 # Attempt to reload - but fail gracefully.
865 try:
866 from importlib import reload
867 sys.path.insert(0,os.getcwd())
868 mod = reload(mod)
869 except: pass
870 # Inspect content of the supplied shared library. Return all methods or throw an error.
871 try: AllModules = jsonable_encoder(Py2X.show(mod,Attribute))
872 except: AllModules = JSONResponse(status_code=404,content={"message": "No such combination exists %s" % ".".join([ModuleID or default,Attribute])},)
873 return AllModules
874
875 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","gitlab","projects"]), tags=[str(self.__pyx_professional)], include_in_schema=False,
876 description=
877 """
878 API for GitLab group service. List all projects inside a given group.
879 """)
880 def api_gitlab_projects(
881 Token: SecretStr =Query(...,
882 description=
883 """
884 GitLab X-API-Token valid for the group.
885 """
886 ),
887 GroupID: Optional[str] = Query("6371",
888 description=
889 """
890 The group identifier. Defaults to group STMLab.
891 """
892 )):
893 """
894 API to fetch information about all projects within a group.
895 """
896 # Import GitLab API interface from minimum working example
897 from PyXMake.VTL import gitlab #@UnresolvedImport
898 # Fetch URL and HEADER
899 url, header = gitlab.datacheck(token=Token.get_secret_value())
900 # Execute command
901 r = requests.get(posixpath.join(url,"groups",str(GroupID),"projects"), headers= header)
902 # Return a dictionary of projects with their id.
903 return JSONResponse(json.dumps({x["name"]:x["id"] for x in r.json()}))
904
905 @self.Router.post(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","gitlab","groups"]), tags=[str(self.__pyx_professional)], include_in_schema=False,
906 description=
907 """
908 API for GitLab. List all groups avoiding decoding error. A dictionary for filtering the results can be provided.
909 """)
910 def api_gitlab_groups(
911 Token: SecretStr =Query(...,
912 description=
913 """
914 GitLab X-API-Token valid for the instance.
915 """
916 ),
917 Filter: Optional[dict] = Body({"name":"STMLab"},
918 description=
919 """
920 Filtering results for key, value pairs. Defaults to filtering for group STMLab.
921 """
922 )):
923 """
924 API to fetch information about all groups within an instance.
925 """
926 import ast
927 # Import GitLab API interface from minimum working example
928 from PyXMake.VTL import gitlab #@UnresolvedImport
929 # Fetch URL and HEADER
930 url, header = gitlab.datacheck(token=Token.get_secret_value())
931 # Default variables
932 status_code = 500; result = {}
933 try:
934 r = requests.get(posixpath.join(url,"groups?per_page=100"), headers=header)
935 # Avoid encoding or decoding errors.
936 encoded_valid = str(r.content).encode("ascii","ignore").decode()
937 decoded_valid = json.loads(ast.literal_eval(encoded_valid).decode("ascii",errors="ignore"))
938 status_code = r.status_code; result = decoded_valid
939 # Apply filter (if given)
940 if Filter: result = [x for x in result if all([x[key] in [value] for key, value in Filter.items()])]
941 except: pass
942 # Return response if successful. An empty dictionary and an error code otherwise.
943 return JSONResponse(status_code=status_code,content=result,)
944
945 @self.Router.post(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","gitlab","packages"]), tags=[str(self.__pyx_professional)], include_in_schema=False,
946 description=
947 """
948 API for GitLab group service. List all packages available within a given group.
949 """)
950 def api_gitlab_packages(
951 Token: SecretStr =Query(...,
952 description=
953 """
954 GitLab X-API-Token valid for the group.
955 """
956 ),
957 GroupID: Optional[str] = Query("6371",
958 description=
959 """
960 The group identifier. Defaults to group STMLab.
961 """
962 ),
963 Filter: Optional[dict] = Body({"package_type":"generic","name":"STMLab"},
964 description=
965 """
966 Filtering results for key, value pairs. Defaults to filtering for the installer of STMLab.
967 """
968 )):
969 """
970 API to fetch information about all packages with a group sorted by their version.
971 """
972 import operator
973 # Import GitLab API interface from minimum working example
974 from PyXMake.VTL import gitlab #@UnresolvedImport
975 # Fetch URL and HEADER
976 url, header = gitlab.datacheck(token=Token.get_secret_value())
977 # Procedure
978 status_code = 500; result = {}
979 try:
980 r = requests.get(posixpath.join(url,"groups",str(GroupID),"packages"), params={"exclude_subgroups":False}, headers= header)
981 unsorted = [x for x in r.json() if all([x[key] in [value] for key, value in Filter.items()])]
982 result =sorted(unsorted, key=operator.itemgetter("version"))[::-1]
983 if result: status_code = 200;
984 except: pass
985 # Return response if successful. An empty dictionary and an error code otherwise.
986 return JSONResponse(status_code=status_code,content=result,)
987
988 @self.Router.put(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","gitlab","job"]), tags=[str(self.__pyx_professional)], include_in_schema=False,
989 description=
990 """
991 API for GitLab job service. Run a given project pipeline job with non-default environment variables.
992 """)
993 def api_gitlab_job(
994 Token: SecretStr = Query(...,
995 description=
996 """
997 GitLab X-API-Token valid for the project.
998 """
999 ),
1000 ProjectID: Optional[str] = Query(str(12702),
1001 description=
1002 """
1003 The project identifier. Defaults to PyXMake's CI pipeline.
1004 """
1005 ),
1006 Job: Optional[str] = Query("pyx_core_docs",
1007 description=
1008 """
1009 A job name. Defaults to a pipeline job for documenting PyXMake.
1010 """
1011 )):
1012 """
1013 API to run a GitLab CI job with non-default environment variables.
1014 """
1015 # Import GitLab API interface from minimum working example
1016 from PyXMake.VTL import gitlab #@UnresolvedImport
1017 # Execute the command and return the result
1018 return JSONResponse(json.dumps(gitlab.pipeline(Token.get_secret_value(), ProjectID, job_name=Job)))
1019
1020 @self.Router.put(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","gitlab","pipeline"]), tags=[str(self.__pyx_professional)], include_in_schema=False,
1021 description=
1022 """
1023 API for GitLab pipeline service. Run a given project pipeline with non-default environment variables.
1024 """)
1025 def api_gitlab_pipeline(
1026 Token: SecretStr = Query(...,
1027 description=
1028 """
1029 GitLab X-API-Token valid for the project.
1030 """
1031 ),
1032 ProjectID: Optional[str] = Query(str(12702),
1033 description=
1034 """
1035 The project identifier. Defaults to PyXMake's CI pipeline.
1036 """
1037 )):
1038 """
1039 API to run a GitLab CI pipeline.
1040 """
1041 # Import GitLab API interface from minimum working example
1042 from PyXMake.VTL import gitlab #@UnresolvedImport
1043 # Execute the command and return the result
1044 return JSONResponse(json.dumps(gitlab.pipeline(Token.get_secret_value(), ProjectID)))
1045
1046 @self.Router.api_route(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","gitlab","variable","{kind}"]),
1047 methods=["GET", "PUT","POST", "PATCH","DELETE","HEAD"],
1048 tags=[str(self.__pyx_professional)], include_in_schema=False,
1049 openapi_extra={"requestBody": {"content": {"application/json": {},},"required": False},},
1050 description=
1051 """
1052 API for GitLab pipeline service. Interact with a given group-level variable.
1053 """)
1054 async def api_gitlab_variable(
1055 request: Request,
1056 kind: Enum("VariableObjectKind",{"groups":"group","projects":"project"}),
1057 Token: SecretStr = Query(...,
1058 description=
1059 """
1060 GitLab X-API-Token valid for the call.
1061 """
1062 ),
1063 ID: str = Query(str(6371),
1064 description=
1065 """
1066 The project or group identifier. Defaults to group STMLab.
1067 """
1068 )):
1069 """
1070 API shim to interact with GitLab variables on both project and group level.
1071 """
1072 # Import GitLab API interface from minimum working example
1073 from PyXMake.VTL import gitlab #@UnresolvedImport
1074 # Fetch URL and HEADER
1075 url, header = gitlab.datacheck(token=Token.get_secret_value())
1076 # Get method from current request
1077 method = str(request.method).lower()
1078 # Default variables
1079 body=b""; headers={"Content-Type":"application/json"}; result = {}; status_code = 500;
1080 try:
1081 # Obtain request body. This always exists - so always "checking" works just fine.
1082 body = await request.body()
1083 # Forward the request
1084 if body: body = json.loads(body)
1085 url = posixpath.join(url,str(kind.name), ID,"variables");
1086 # Add path parameter
1087 if isinstance(body,dict) and not method in ["post"]: url = posixpath.join(url,body.get("key",""));
1088 # Call API
1089 r = requests.request(method, url, params=request.query_params, data=body, headers=header)
1090 # Collect the result
1091 status_code = r.status_code; result = r.content; headers.update(r.headers)
1092 try: headers.pop("Content-Encoding") # Remove encoding
1093 except: pass
1094 except: pass
1095 # Return the forwarded response
1096 return Response(status_code=status_code, content=result, headers=headers)
1097
1098 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","gitlab","runner"]), tags=[str(self.__pyx_professional)], include_in_schema=False,
1099 description=
1100 """
1101 API for GitLab service. Creates a GitLab runner dedicated for the given project registration token on FA-Services.
1102 """)
1103 def api_gitlab_runner(
1104 Registration: str = Query(...,
1105 description=
1106 """
1107 GitLab registration token (mandatory).
1108 """
1109 ),
1110 Token: SecretStr = Query("",
1111 description=
1112 """
1113 Portainer X-API-Token.
1114 """
1115 )):
1116 """
1117 API to create a GitLab runner dedicated for the given project registration token.
1118 """
1119 # Local variables
1120 delimn = "_";
1121 # Default values
1122 is_service = True; with_tmpfs = True; with_persistent = True
1123 tmpfs = []; mounts = []; result = {}; status_code = 500;
1124 try:
1125 from PyXMake.VTL import portainer #@UnresolvedImport
1126 from PyCODAC.API.Remote import Server
1127 # Procedure
1128 runner = str(delimn.join(["stmlab_runner",str(uuid.uuid4())]))
1129 umlshm = runner.split(delimn); umlshm.insert(2,"umlshm"); umlshm = str(delimn.join(umlshm))
1130 persistent = runner.split(delimn); persistent.insert(2,"persistent"); persistent = str(delimn.join(persistent))
1131 # Fetch base url and login information from the given token
1132 url, header = portainer.main(str(getattr(Token,"get_secret_value",str)()) or Server.auth['X-API-KEY'], datacheck=True);
1133 # Get any valid docker agent and node
1134 node = str([x["Id"] for x in requests.get(posixpath.join(url,"endpoints"), headers=header).json()][-1])
1135 agent = list(random.choice(Server.GetNodeID()).keys())[0]; header.update({"X-PortainerAgent-Target":agent})
1136 # Add fast scratch storage and persistent docker storage (if requested)
1137 if (with_tmpfs or with_persistent) and is_service:
1138 mounts.extend([{"ReadOnly": False,"Source": umlshm,"Target": "/umlshm", "Type":"volume"}])
1139 mounts.extend([{"ReadOnly": False,"Source": persistent,"Target": "/persistent", "Type":"volume"}])
1140 # Both mounts are temporary
1141 if mounts and (with_tmpfs and not with_persistent): tmpfs = [umlshm, persistent]
1142 elif mounts and (with_tmpfs and with_persistent): tmpfs = [umlshm]
1143 # Create a new shared memory volume with the host for each instance
1144 for x in tmpfs:
1145 data = {"Name": x,"DriverOpts": {"device": "tmpfs","o": "rw,nosuid,nodev,exec","type": "tmpfs"}}
1146 # Create a new associated shared storage volume for each instance.
1147 r = requests.post(posixpath.join(url,"endpoints",node,"docker","volumes","create"),
1148 json=data, headers=header); r.raise_for_status()
1149 # Create a persistent volume. Only meaningful when a service is created.
1150 if with_persistent and is_service:
1151 # Create a volume in the process
1152 r = requests.post(posixpath.join(url,"endpoints",node,"docker","volumes","create"),
1153 json={"Name":persistent}, headers=header); r.raise_for_status()
1154 # Create a new GitLab runner with predefined properties. Defined as a service definition
1155 data = {
1156 "Name": runner,"TaskTemplate": {
1157 "Placement": {"Constraints": ["node.hostname==%s" % agent]},
1158 "ContainerSpec": {"Image": "harbor.fa-services.intra.dlr.de/stmlab/orchestra:latest","Mounts": mounts,
1159 "Args": ["./runner.sh","--token=%s" % str(Registration),"--return=false","--locked=true"],
1160 "Env": ["DISK=80G","MEM=8G"], "Mode": {"Replicated": {"Replicas": 1}}}}
1161 }
1162 # Create as a new service
1163 if is_service:
1164 r = requests.post(posixpath.join(url,"endpoints",node,"docker","services","create"), json=data, headers=header);
1165 result = r.json()
1166 else:
1167 ## Just create a container instance (more stable)
1168 # Apply changes
1169 data = data["TaskTemplate"].get('ContainerSpec'); data["Cmd"] = data.pop("Args");
1170 r = requests.post(posixpath.join(url,"endpoints",node,"docker","containers","create"),
1171 params={"name":runner}, json=data, headers=header); r.raise_for_status()
1172 # Actually start the container.
1173 r = requests.post(posixpath.join(url,"endpoints",node,"docker","containers",r.json()["Id"],"start"), headers=header);
1174 if r.status_code in [204,304,404]: result = "Success"
1175 if result: status_code = 200
1176 # Always exit gracefully
1177 except Exception: pass
1178 # Return a dictionary of projects with their id.
1179 return JSONResponse(status_code=status_code,content=result,)
1180
1181 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","gitlab","user"]), tags=[str(self.__pyx_professional)], include_in_schema=False,
1182 description=
1183 """
1184 API for GitLab service. Returns the current user for a given API token.
1185 """)
1186 def api_gitlab_user(
1187 Token: SecretStr = Query(...,
1188 description=
1189 """
1190 GitLab X-API-Token.
1191 """
1192 )):
1193 """
1194 API to get the current active user from an API Token.
1195 """
1196 # Import GitLab API interface from minimum working example
1197 from PyXMake.VTL import gitlab #@UnresolvedImport
1198 # Fetch URL and HEADER
1199 url, header = gitlab.main(token=Token.get_secret_value(), datacheck=True)
1200 # Default values
1201 result = {}; status_code = 500;
1202 try:
1203 r = requests.get(posixpath.join(url,"user"), headers= header); result = r.json()
1204 if result: status_code = r.status_code
1205 except: pass
1206 # Return a dictionary of projects with their id.
1207 return JSONResponse(status_code=status_code,content=result,)
1208
1209 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","portainer","user"]), tags=[str(self.__pyx_professional)], include_in_schema=False,
1210 description=
1211 """
1212 API for Portainer service. Returns the current user for a given API token.
1213 """)
1214 def api_portainer_user(
1215 Token: SecretStr = Query(...,
1216 description=
1217 """
1218 Portainer X-API-Token.
1219 """
1220 )):
1221 """
1222 API to get the current active user from an API Token.
1223 """
1224 # Default variables
1225 status_code = 500; result = []
1226 # Import Portainer API interface from minimum working example
1227 from PyXMake.VTL import portainer #@UnresolvedImport
1228 # Execute the command and return the result
1229 try:
1230 result = portainer.main(Token.get_secret_value());
1231 # Result must be a list with one element.
1232 if isinstance(result, list):
1233 status_code = 200; result = result[0]
1234 # The token is invalid and result contains a message.
1235 else: status_code = 401
1236 except: pass
1237 # Return the current active user.
1238 return JSONResponse(status_code=status_code,content=result,)
1239
1240 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","shepard","user"]), tags=[str(self.__pyx_professional)], include_in_schema=False,
1241 description=
1242 """
1243 API for Shepard service. Returns the current user for a given API token.
1244 """)
1245 def api_shepard_user(
1246 Token: SecretStr = Query(...,
1247 description=
1248 """
1249 Shepard X-API-Token.
1250 """
1251 )):
1252 """
1253 API to get the current active user from an API Token.
1254 """
1255 # Default variables
1256 status_code = 500; result = {}
1257 try:
1258 # Import Shepard API interface from minimum working example
1259 from PyCODAC.Tools.IOHandling import Shepard #@UnresolvedImport
1260 except: return JSONResponse(status_code=404,content="Service not available",)
1261 # Execute the command and return the result
1262 try:
1263 result = Shepard.GetUser(header= {'X-API-KEY':Token.get_secret_value()}, full=True);
1264 # The token is valid and result contains the user.
1265 if result: status_code = 200;
1266 except: pass
1267 # Return the current active user.
1268 return JSONResponse(status_code=status_code,content=result,)
1269
1270 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","shepard","{kind}"]), tags=[str(self.__pyx_professional)], include_in_schema=False,
1271 description=
1272 """
1273 API for PyXMake. Browse through all publicly available Shepard identifiers present within the scope of this API.
1274 """ )
1275 def api_shepard_show(
1276 kind: api_database()[0],
1277 ID: Optional[str] = Query(None,
1278 description=
1279 """
1280 An user-defined container ID.
1281 """
1282 ),
1283 Token: SecretStr = Query(None,
1284 description=
1285 """
1286 Shepard X-API-Token.
1287 """
1288 )):
1289 """
1290 API to connect to all available data identifiers using PyXMake.
1291 """
1292 # Default variables
1293 result = []; status_code = 500; delimn = "_";
1294 DB = ID or str(kind.name); options = api_database()[-1]
1295 try:
1296 from PyCODAC.Tools.IOHandling import Shepard #@UnresolvedImport
1297 # Try non-default Shepard Token
1298 try: Shepard.SetHeader( {'X-API-KEY':Token.get_secret_value()} )
1299 except: pass
1300 finally: Shepard.GetContainerID(DB) # Check if file database can be accessed.
1301 except: return JSONResponse(status_code=404,content="Service not available",)
1302 # Connect to an accessible data browser.
1303 try:
1304 # Collect all public data
1305 vault = list(Utility.ArbitraryFlattening([ Shepard.GetContent(DB, query=options[str(kind.value)], latest=False) ]))
1306 result.extend([{ next(iter(x)) : {"id":x[next(iter(x))][0],"created":x[list(x.keys())[0]][-1],"parent":DB.split(delimn)[-1].lower()}} for x in vault])
1307 status_code = 200 # Result can be empty, but the command has succeeded anyways.
1308 except ConnectionRefusedError:
1309 # Do not consider an empty time series as an error
1310 if options[str(kind.value)] in ["timeseries"]: status_code = 200
1311 except:
1312 # If an ID is explicitly given and cannot be reached.
1313 if ID: return JSONResponse(status_code=404,content="The given ID '%s' cannot be resolved for kind '%s'" % (ID,options,options[str(kind.value)]),)
1314 pass # Fail gracefully
1315 finally:
1316 result = list(Utility.ArbitraryFlattening(result))
1317 if result: status_code = 200
1318 # Present result to FAST API
1319 return JSONResponse(status_code=status_code, content=result,)
1320
1321 @self.Router.delete(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","shepard","{kind}"]), tags=[str(self.__pyx_professional)], include_in_schema=False,
1322 description=
1323 """
1324 API for PyXMake. Delete an available data object identifier present within the scope of this API.
1325 """ )
1326 def api_shepard_delete(
1327 kind: api_database()[0],
1328 OID: Optional[str] = Query(...,
1329 description=
1330 """
1331 The object identifier of the data.
1332 """
1333 ),
1334 ID: Optional[str] = Query(None,
1335 description=
1336 """
1337 An user-defined container ID.
1338 """
1339 ),
1340 Token: SecretStr = Query(None,
1341 description=
1342 """
1343 Shepard X-API-Token.
1344 """
1345 )):
1346 """
1347 API to delete any available data identifier using PyXMake.
1348 """
1349 # Default variables
1350 result = []; status_code = 500; DB = ID or str(kind.name); options = api_database()[-1]
1351 try:
1352 from PyCODAC.Tools.IOHandling import Shepard #@UnresolvedImport
1353 # Try non-default Shepard Token
1354 try: Shepard.SetHeader( {'X-API-KEY':Token.get_secret_value()} )
1355 except: pass
1356 finally: Shepard.GetContainerID(DB) # Check if file database can be accessed.
1357 except: return JSONResponse(status_code=404,content="Service not available",)
1358 # Delete the requested object.
1359 try:
1360 Shepard.DeleteContent(DB, OID, query=options[str(kind.value)], latest=False)
1361 status_code = 200;
1362 except:
1363 # If an ID is explicitly given and cannot be reached.
1364 if ID: return JSONResponse(status_code=404,content="The given ID '%s' cannot be resolved for kind '%s'" % (ID,options,options[str(kind.value)]),)
1365 finally: result = "Success"
1366 # Present result to FAST API
1367 return JSONResponse(status_code=status_code, content=result)
1368
1369 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","overleaf","session"]), tags=[str(self.__pyx_professional)], include_in_schema=False,
1370 description=
1371 """
1372 API for Overleaf service. Creates a valid API session token for Overleaf to automate tasks.
1373 """)
1374 def api_overleaf_session(
1375 Token: SecretStr = Query(...,
1376 description=
1377 """
1378 Credentials given as Overleaf X-API-Token with its value set to the result of:
1379 "echo -n "<'YourUserName'>:<'YourPasswort'>" | base64"
1380 """
1381 ),
1382 base_url: Optional[str] = Query("", include_in_schema=False,
1383 description=
1384 """
1385 Base URL of the Overleaf instance in question. Defaults to Overleaf running on FA-Services.
1386 """)):
1387 """
1388 Creates a valid API session token for Overleaf.
1389 """
1390 # Default variables
1391 status_code = 500; result = {}
1392 # Import Overleaf API interface from minimum working example
1393 from PyXMake.Build.Make import Latex #@UnresolvedImport
1394 # Execute the command and return the result
1395 try: status_code, result = Latex.session(*base64.b64decode(Token.get_secret_value()).decode('utf-8').split(":",1), base_url=base_url, use_cache=False);
1396 except: pass
1397 # Return the current active user.
1398 return JSONResponse(status_code=status_code,content=result,)
1399
1400 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","overleaf","show"]), tags=[str(self.__pyx_professional)], include_in_schema=False,
1401 description=
1402 """
1403 API for Overleaf service. Compile an Overleaf project remotely and return all files.
1404 """)
1405 def api_overleaf_show(
1406 Token: SecretStr = Query(...,
1407 description=
1408 """
1409 Credentials given as Overleaf X-API-Token with its value set to the result of:
1410 "echo -n "<'YourUserName'>:<'YourPasswort'>" | base64"
1411 """
1412 ),
1413 ProjectID: str = Query(..., description=
1414 """
1415 An Overleaf Project ID.
1416 """
1417 ),
1418 base_url: Optional[str] = Query("", include_in_schema=False, description=
1419 """
1420 Base URL of the Overleaf instance in question. Defaults to Overleaf running on FA-Services.
1421 """)):
1422 """
1423 Compile an Overleaf project remotely and return all files.
1424 """
1425 # Default variables
1426 status_code = 500; result = {}
1427 # Import Overleaf API interface from minimum working example
1428 from PyXMake.Build.Make import Latex #@UnresolvedImport
1429 # Execute the command and return the result
1430 try:
1431 result = Latex.show(ProjectID, *base64.b64decode(Token.get_secret_value()).decode('utf-8').split(":",1), base_url=base_url, use_cache=False) ;
1432 if result: status_code = 200 ;
1433 except: pass
1434 # Return the current active user.
1435 return JSONResponse(status_code=status_code,content=result,)
1436
1437 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","overleaf","download","{kind}"]), tags=[str(self.__pyx_professional)], include_in_schema=False,
1438 description=
1439 """
1440 API for Overleaf service. Compile an Overleaf project remotely and return all files.
1441 """)
1442 def api_overleaf_download(
1443 kind: Enum("DataObjectKind",{"zip":"zip","pdf":"pdf"}),
1444 Token: SecretStr = Query(...,
1445 description=
1446 """
1447 Credentials given as Overleaf X-API-Token with its value set to the result of:
1448 "echo -n "<'YourUserName'>:<'YourPasswort'>" | base64"
1449 """
1450 ),
1451 ProjectID: str = Query(..., description=
1452 """
1453 An Overleaf Project ID.
1454 """
1455 ),
1456 base_url: Optional[str] = Query("", include_in_schema=False, description=
1457 """
1458 Base URL of the Overleaf instance in question. Defaults to Overleaf running on FA-Services.
1459 """)):
1460 """
1461 Compile an Overleaf project remotely and return all files.
1462 """
1463 # Import Overleaf API interface from minimum working example
1464 from PyXMake.Build.Make import Latex #@UnresolvedImport
1465 # Execute the command and return the result
1466 try:
1467 with Utility.ChangedWorkingDirectory(VTL.Scratch):
1468 FilePath = Latex.download(ProjectID, *base64.b64decode(Token.get_secret_value()).decode('utf-8').split(":",1), output_format=str(kind.value), base_url=base_url, use_cache=False) ;
1469 if not os.path.exists(FilePath): raise FileNotFoundError
1470 except: return JSONResponse(status_code=404,content="Service not available",)
1471 # Present result to FAST API
1472 return FileResponse(path=FilePath,filename=Utility.PathLeaf(FilePath))
1473
1474 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","time"]), tags=[str(self.__pyx_professional)], include_in_schema=False,
1475 description=
1476 """
1477 API for time service. Get the current system time of this API in ISO 8601 format (UTC).
1478 """ )
1479 def api_time_show(request: Request):
1480 """
1481 Get the current system time of this API in ISO 8601 format (UTC).
1482 """
1483 # Initialize everything
1484 timestamp = "Unknown system time"
1485 # Construct the response
1486 try:
1487 timestamp = str(datetime.datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ"))
1488 status_code = 200
1489 except: status_code = 500
1490 # Return current time in ISO 8601 format (UTC).
1491 return JSONResponse(status_code=status_code,content=timestamp,)
1492
1493 @self.Router.post(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","time"]), tags=[str(self.__pyx_professional)], include_in_schema=False,
1494 description=
1495 """
1496 API for time service. Return the given time from any time format in UTC format (nanoseconds).
1497 """ )
1498 def api_time_inspect(request: Request,
1499 Time: str = Query(...,
1500 description=
1501 """
1502 Arbitrary valid time string. The string is parsed into an internal date object.
1503 """
1504 )):
1505 """
1506 Return the given time from any time format in UTC format (nanoseconds).
1507 """
1508 from dateutil.parser import parse
1509 # Initialize everything
1510 timestamp = "Unknown input time"
1511 # Construct the response
1512 try:
1513 timestamp = int(parse(Time).replace(tzinfo=datetime.timezone.utc).timestamp() * 1e9)
1514 status_code = 200
1515 except: status_code = 500
1516 # Return current time in UTC format (nanoseconds).
1517 return JSONResponse(status_code=status_code,content=timestamp,)
1518
1519 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","secret"]), tags=[str(self.__pyx_professional)],
1520 description=
1521 """
1522 API for PyXMake. Show a list of all available secrets for a given token.
1523 """)
1524 def api_secret_show(request: Request,
1525 Token: SecretStr = Query(...,
1526 description=
1527 """
1528 Portainer X-API-Token. The token is stored in your personal user space.
1529 """
1530 )):
1531 """
1532 API to get the current active user from an API Token.
1533 """
1534 try:
1535 from PyXMake.VTL import portainer #@UnresolvedImport
1536 result = portainer.main(Token.get_secret_value());
1537 if not isinstance(result, list): raise ValueError
1538 except: return JSONResponse(status_code=404,content="Service not available",)
1539 # Operate fully in a temporary directory
1540 with Utility.TemporaryDirectory():
1541 # Fetch header and base address
1542 url, header = portainer.main(Token.get_secret_value(), datacheck=True);
1543 # Create a new secret
1544 secret_url = posixpath.join(url,"endpoints",str([x["Id"] for x in requests.get(posixpath.join(url,"endpoints"), headers=header).json()][-1]),"docker","secrets")
1545 r = requests.get(secret_url, headers=header)
1546 # Return the current active user.
1547 return JSONResponse(status_code=r.status_code,content=r.json(),)
1548
1549 @self.Router.post(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","secret"]), tags=[str(self.__pyx_professional)],
1550 description=
1551 """
1552 API for PyXMake. Create a new secret for the given token.
1553 """)
1554 def api_secret_upload(request: Request,
1555 payload: UploadFile = File(...,
1556 description=
1557 """
1558 Binary data stream.
1559 """
1560 ),
1561 Token: SecretStr = Query(...,
1562 description=
1563 """
1564 Portainer X-API-Token. The token is stored in your personal user space.
1565 """
1566 ),
1567 SecretID: str = Query(None,
1568 description=
1569 """
1570 An optional name for the secret. Defaults to the name of the given file.
1571 """
1572 )):
1573 """
1574 API to get the current active user from an API Token.
1575 """
1576 try:
1577 from PyXMake.VTL import portainer #@UnresolvedImport
1578 result = portainer.main(Token.get_secret_value());
1579 if not isinstance(result, list): raise ValueError
1580 except: return JSONResponse(status_code=404,content="Service not available",)
1581 # Operate fully in a temporary directory
1582 with Utility.TemporaryDirectory():
1583 # Collapse data block in current workspace
1584 with open(payload.filename,'wb+') as f: f.write(payload.file.read())
1585 with io.open(payload.filename,'r',encoding='utf-8')as f: content = f.read()
1586 # Create a base64 encoded string
1587 content = requests.post(str(request.url.scheme)+"://%s/2/PyXMake/api/encoding/base64" % self.APIBase, params={"message":content}, verify=False).json()
1588 url, header = portainer.main(Token.get_secret_value(), datacheck=True);
1589 # Assemble request body
1590 body = { 'Data': content,'Name':SecretID or payload.filename,'Labels': None } ;
1591 # Create a new secret
1592 secret_url = posixpath.join(url,"endpoints",
1593 str([x["Id"] for x in requests.get(posixpath.join(url,"endpoints"), headers=header).json()][-1]),"docker","secrets","create")
1594 r = requests.post(secret_url, json=body, headers=header)
1595 # Return the current active user.
1596 return JSONResponse(status_code=r.status_code,content=r.json(),)
1597
1598 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","file","{kind}"]), tags=[str(self.__pyx_professional)],
1599 description=
1600 """
1601 API for PyXMake. Browse through all available file identifiers present within the scope of this API.
1602 """ )
1603 def api_file_show(request: Request, kind: FileManager):
1604 """
1605 API to connect to all available file identifiers using PyXMake.
1606 """
1607 # Default variables
1608 result = []; status_code = 500;
1609 try:
1610 from PyCODAC.VTL import Fetch #@UnresolvedImport
1611 except: return JSONResponse(status_code=404,content="Service not available",)
1612 # Initialize random seed generator
1613 rd = random.Random();
1614 # Connect to an accessible file browser.
1615 try:
1616 # Collect all public data
1617 if str(kind.value) in ["all","public"]:
1618 result = requests.get(str(request.url.scheme)+"://%s/2/PyXMake/api/shepard/file" % self.APIBase, verify=False).json()
1619 # Collect all private data
1620 if kind.value in ["all","private"]:
1621 private = [];
1622 for root, _, files in Utility.PathWalk(Fetch, startswith=(".","__")):
1623 if files and Utility.PathLeaf(root).split("_")[0].isnumeric():
1624 private.extend([{x:{"parent":Utility.PathLeaf(root),"created":time.ctime(os.path.getctime(os.path.join(root,x)))}} for x in files])
1625 # Add an unique identifier for compatibility
1626 for i,x in enumerate(private):
1627 # Create a pseudo-random UUID
1628 rd.seed(int(i)); x[next(iter(x))].update({"id": str(uuid.UUID(int=rd.getrandbits(128), version=4))})
1629 result.append(x)
1630 except: pass # Fail gracefully
1631 finally:
1632 result = list(Utility.ArbitraryFlattening(result))
1633 if result: status_code = 200
1634 # Present result to FAST API
1635 return JSONResponse(status_code=status_code, content=result,)
1636
1637 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","file"]), tags=[str(self.__pyx_professional)],
1638 description=
1639 """
1640 API for PyXMake. Download a file from an available data source by its id.
1641 """ )
1642 def api_file_download(request: Request, FileID: str = Query(...)):
1643 """
1644 API to upload a given file using PyXMake.
1645 """
1646 # Default variables
1647 FileDB = [data.name for data in api_database()[0] if data.value == Utility.PathLeaf(str(request.url).split("?")[0])][0]
1648 try:
1649 from PyCODAC.VTL import Fetch #@UnresolvedImport
1650 from PyCODAC.Tools.IOHandling import Shepard #@UnresolvedImport
1651 Shepard.GetContainerID(FileDB) # Check if file database can be accessed.
1652 except: return JSONResponse(status_code=404,content="Service not available",)
1653 # Test for data consistency
1654 try:
1655 # Fetch content from other service
1656 AllFiles = requests.get(str(request.url.scheme)+"://%s/2/PyXMake/api/file/all" % self.APIBase, verify=False).json()
1657 ListofIDs = [x[next(iter(x))].get("id") for x in AllFiles]
1658 if len(ListofIDs) != len(set(ListofIDs)): raise IndexError # List are not of equal length: There are duplicates, which should never happen.
1659 except: return JSONResponse(status_code=500,content="Internal database error: There are multiple files sharing the same ID.",)
1660 # Check if the given ID is valid. Return if False.
1661 if not FileID in ListofIDs: return JSONResponse(status_code=404,content="File not found",)
1662 # Operate fully in a temporary directory
1663 with Utility.TemporaryDirectory():
1664 # Collect meta data of the file in question. Required properties are its id, parent and name.
1665 data = [(x[next(iter(x))].get("id"),x[next(iter(x))].get("parent"),next(iter(x))) for x in AllFiles if FileID == x[next(iter(x))].get("id") ][0]
1666 try:
1667 # Try to download the file from internal vault. Allow multiple files sharing the same name.
1668 if FileDB.split("_")[-1].lower() in [data[1]]: FilePath = Shepard.DownloadFile(FileDB, FileID, latest=False)[0]
1669 # An internal resource is requested. Provide a reference to its path.
1670 else: FilePath = os.path.join(Fetch,data[1],data[-1])
1671 except IndexError: return JSONResponse(status_code=500,content="Internal database error: Permission denied.",)
1672 # Present result to FAST API
1673 return FileResponse(path=FilePath,filename=Utility.PathLeaf(FilePath))
1674
1675 @self.Router.post(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","file","{kind}"]), tags=[str(self.__pyx_professional)],
1676 description=
1677 """
1678 API for PyXMake. Connects an user with an available file browser and uploads the given data.
1679 """ )
1680 def api_file_upload(request: Request,
1681 kind: Enum("ScopeObjectKind",{"local":"private","shared":"public"}),
1682 payload: UploadFile = File(...,
1683 description=
1684 """
1685 Binary data stream.
1686 """
1687 )):
1688 """
1689 API to upload a given file using PyXMake.
1690 """
1691 import functools
1692 # Default variables
1693 status_code = 500;
1694 FileDB = [data.name for data in api_database()[0] if data.value ==Utility.PathLeaf(posixpath.dirname(str(request.url)))][0]
1695 try:
1696 if str(kind.value) in ["public"]:
1697 from PyCODAC.Tools.IOHandling import Shepard #@UnresolvedImport
1698 Shepard.GetContainerID(FileDB) # Check if file database can be accessed.
1699 elif str(kind.value) in ["private"]:
1700 from PyCODAC.Database import MatDB
1701 except ValueError: return JSONResponse(status_code=404,content="Service not available",)
1702 # Operate fully in a temporary directory
1703 with Utility.TemporaryDirectory():
1704 # Refer to MongoDB explicitly
1705 from PyCODAC.Database import Mongo
1706 # Collapse data block in current workspace
1707 with open(payload.filename,'wb+') as f: f.write(payload.file.read())
1708 # Upload the given file, including large file support.
1709 if str(kind.value) in ["public"]:
1710 Shepard.UploadFile(FileDB, payload.filename, os.getcwd(), unique=False)
1711 Shepard.UpdateContainer(FileDB, collections=Mongo.base_collection, latest=False)
1712 # Return OID for the user
1713 result = Shepard.GetContent(FileDB)[payload.filename]
1714 elif str(kind.value) in ["private"]:
1715 # Store data inaccessible w/o its explicit id
1716 url = functools.reduce(lambda a, kv: a.replace(*kv), (("eoddatamodel","filemanager"),("materials","files"),),MatDB.base_url)
1717 result = [Utility.FileUpload(posixpath.join(url,""), payload.filename, None, verify=False).json()["file_id"],
1718 datetime.datetime.fromtimestamp(time.time()).strftime('%Y-%m-%dT%H:%M:%SZ')]
1719 # This should never happen.
1720 else: raise NotImplementedError
1721 if result: status_code = 200
1722 # Present result to FAST API
1723 return JSONResponse(status_code=status_code, content=result,)
1724
1725 @self.Router.patch(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","file","format","{kind}"]), tags=[str(self.__pyx_professional)],
1726 description=
1727 """
1728 API for PyXMake. Connects an user with various file formatting options present within the scope of this API.
1729 """ )
1730 def api_file_converter(
1731 kind: Archive,
1732 OutputID: Enum("FormatObjectKind",{"xls":"xls"}),
1733 SourceID: Optional[str] = Query("Settings.xlsx",
1734 description=
1735 """
1736 The name of the main file present in the supplied ZIP archive. The requested output format is given by OutputID.
1737 """),
1738 ZIP: UploadFile = File(...,
1739 description=
1740 """
1741 An compressed archive containing all additional files required for this job. This archive is updated and returned. All input files are preserved.
1742 It additionally acts as an on-demand temporary user directory. It prohibits access from other users preventing accidental interference.
1743 However, the scratch workspace is cleaned constantly. So please download your result immediately.
1744 """
1745 )):
1746 """
1747 API to connect to various file formatting options using PyXMake.
1748 """
1749 # Check if an old file is present in the current workspace. Delete it.
1750 if os.path.exists(os.path.join(VTL.Scratch,ZIP.filename)): os.remove(os.path.join(VTL.Scratch,ZIP.filename))
1751
1752 # Everything is done within a temporary directory. Update the uploaded ZIP folder with new content. Preserve the input.
1753 with Utility.TemporaryDirectory(VTL.Scratch), Utility.UpdateZIP(ZIP.filename, ZIP.file, VTL.Scratch, update=True):
1754 # Procedure
1755 with Utility.FileOutput('result.log'):
1756 try:
1757 # Create a legacy XLS file from input. Required for MRO process.
1758 if str(OutputID.name) in ["xls"] : Utility.ConvertExcel(str(SourceID), os.getcwd())
1759 except: print("File conversion error. Could not convert %s into %s format." % (str(SourceID), str(OutputID.value).upper(),) )
1760 pass
1761
1762 # Present result to FAST API
1763 return FileResponse(path=os.path.join(VTL.Scratch,ZIP.filename),filename=ZIP.filename)
1764
1765 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","image","{kind}"]), tags=[str(self.__pyx_professional)],
1766 description=
1767 """
1768 API for PyXMake. Obtain information about an image.
1769 """)
1770 def api_image_show(kind: Enum("ImageObjectKind",{"public":"public","user":"user"}),
1771 request: Request,
1772 image: str = Query(...,
1773 description=
1774 """
1775 Fully qualified name of the image.
1776 """
1777 )):
1778 """
1779 API to obtain image information
1780 """
1781 # Default status code (Not Documented Error)
1782 result = {}
1783 try:
1784 # Add the correct repository to the front.
1785 if kind.name in ["public"] and not image.startswith("docker.io"): search = posixpath.join("docker.io",str(image))
1786 # We cannot determine assume any named registry. The user is responsible
1787 elif kind.name in ["user"]: search = str(image)
1788 else: raise NotImplementedError
1789 # Refer to internal portainer reference.
1790 base_url = str(request.url.scheme)+"://%s/api/portainer" % self.APIBase
1791 # Use any active instance:
1792 r = requests.get(posixpath.join(base_url,"endpoints")); endpoint = [str(x["Id"]) for x in r.json()][-1]
1793 # Verify that the image exists. Download the image if not.
1794 requests.post(posixpath.join(base_url,"endpoints",endpoint,"docker","images","create"), params={"fromImage":search})
1795 # Return all information about the image as JSON.
1796 result.update(requests.get(posixpath.join(base_url,"endpoints",endpoint,"docker","images",image,"json")).json())
1797 # Success
1798 if result: return JSONResponse(status_code=200, content=result)
1799 # Something bad happened...
1800 except: return JSONResponse(status_code=404, content="Could not determine info for image %s. Image probably not in scope." % image)
1801
1802 @self.Router.post(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","encoding","{kind}"]), tags=[str(self.__pyx_professional)],
1803 description=
1804 """
1805 API for PyXMake. Encode a given secret string.
1806 """)
1807 def api_encode_message(kind: Enum("EncodingObjectKind",{"base64":"base64"}),
1808 message: SecretStr = Query(...,
1809 description=
1810 """
1811 Message (string) to be encoded
1812 """
1813 )):
1814 """
1815 API to encode a message
1816 """
1817 # Use internal base64 encryption method
1818 try:
1819 message = message.get_secret_value()
1820 base64_message = Utility.GetDockerEncoding(message, encoding='utf-8')
1821 return JSONResponse(content=base64_message)
1822 except: return JSONResponse(404, content="Encoding failed.")
1823
1824 @self.Router.put(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","notification","{kind}"]), tags=[str(self.__pyx_professional)],
1825 description=
1826 """
1827 API for PyXMake. Connect with a notification service to send arbitrary mail or Mattermost notifications.
1828 """ )
1829 def api_notification(
1830 kind: Enum("NotificationObjectKind",{"mail":"mail","mattermoost":"mattermost"}),
1831 recipient: str = Query(..., description=
1832 """
1833 Qualified mail address or mattermost channel.
1834 """
1835 ),
1836 notification: dict = Body({"subject":"MyFancyHeader","content":"MyFancyMessage"},
1837 description=
1838 """
1839 A dictionary containing the subject line and the content.
1840 """
1841 )):
1842 """
1843 API for PyXMake. Connect with a notification service.
1844 """
1845 import tempfile
1846
1847 from collections import defaultdict #@UnresolvedImport
1848 from PyXMake.Build import Make #@UnresolvedImport
1849 # Default status code (Not Documented Error)
1850 status_code = 500
1851 # Check whether a connection with FlexLM can be established or not.
1852 default_user = "f_testpa"; default_host = "cara.dlr.de"
1853 secrets = os.path.join(Utility.AsDrive("keys"),default_host)
1854 # Check for Mattermost notification first
1855 if kind.value in ["mattermost"]:
1856 try:
1857 # Retain compatibility
1858 payload={"text": notification.pop("text","") or "\n".join(
1859 [notification.pop("subject"),notification.pop("content")]), "username":notification.pop("username",str(kind.value).title())}
1860 # Add all leftover keys
1861 payload.update(notification)
1862 requests.post(recipient, json=payload)
1863 # We executed the command successfully.
1864 return JSONResponse(status_code=200,content="Success",)
1865 except: return JSONResponse(status_code=404,content="%s service not available" % str(kind.value).title(),)
1866 try: default_key = os.path.join(secrets,[x for x in os.listdir(secrets) if x.split(".")[-1] == x][0])
1867 except: return JSONResponse(status_code=404,content="Service not available",)
1868 # Fail gracefully
1869 try:
1870 # Operate fully in a temporary directory
1871 with Utility.TemporaryDirectory():
1872 # Create a temporary file name
1873 tmp = str(next(tempfile._get_candidate_names()))
1874 command = "echo '%s' | mail -s '%s' %s" % (notification["content"], notification["subject"], recipient, )
1875 connect = Make.SSH("Notification",[]); connect.Settings(default_user, key=default_key, host=default_host, timeout=1)
1876 # Execute the SSH script and log the result
1877 with Utility.FileOutput(tmp): Utility.SSHPopen(connect.ssh_client, ntpath.pathsep.join([command]),verbosity=2)
1878 # Read logging information into memory
1879 with open(tmp,"r") as f: content = f.read()
1880 # Success
1881 _ = json.dumps(content); status_code = 200
1882 # Prevent endless loops if service is not reachable
1883 except TimeoutError: return JSONResponse(status_code=404,content="%s service not available" % str(kind.value).title(),)
1884 except: pass
1885 # Just return a success status
1886 return JSONResponse(status_code=status_code,content="Success",)
1887
1888## @class PyXMake.API.Frontend
1889# Class instance to define PyXMake's web API instance
1890class Frontend(Base,Backend):
1891 """
1892 Class instance to define PyXMake's server instance for a web API.
1893 """
1894 # Some immutable variables
1895 __pyx_api_delimn = "/"
1896 __pyx_doc_path = os.path.normpath(os.path.join(os.path.dirname(VTL.__file__),"doc","pyx_core","html"))
1897 __pyx_url_path = __pyx_api_delimn.join(["https:","","fa_sw.pages.gitlab.dlr.de","stmlab",str(PyXMake.__name__),"index.html"])
1898
1899 # These are the captions for the Swagger UI
1900 __pyx_guide = "Guide"
1901 __pyx_interface = "Interface"
1902 __pyx_professional = "Professional"
1903
1904 def __init__(self, *args, **kwargs):
1905 """
1906 Initialization of PyXMake's Frontend API.
1907 """
1908 super(Frontend, self).__init__(*args, **kwargs)
1910
1911 # Collection API meta data information
1912 self.APIMeta = kwargs.get("Meta", {})
1913
1914 # Added API subsection meta data information
1915 self.APITags = [
1916 {"name": self.__pyx_guide,"description": "Operations to obtain server documentation."},
1917 {"name": self.__pyx_interface,"description": "Operations for all users."},
1918 {"name": self.__pyx_professional,"description": "Operations for experienced users and developers."}]
1919
1920 # Added default swagger mode. Defaults to hiding the schema section
1921 self.APISwagger = {"defaultModelsExpandDepth": -1}
1922
1923 # Redefine API to create the correct description
1924 self.APIAPI = FastAPI(title="PyXMake API",
1925 # Use internal package version instead of global package number. Kept for backwards compatibility
1926 version = getattr(PyXMake,"__version__", self.APIMeta.pop("version","1.0.0")),
1927 description ="Simplifying complex compilations",
1928 docs_url = self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"api","documentation"]),
1929 swagger_ui_parameters = self.APISwagger, openapi_tags = self.APITags, **self.APIMeta)
1930
1931 # Define variables for self access
1932 self.APIHost = kwargs.get("HostID",socket.gethostname())
1933 self.APIPort = kwargs.get("PortID",str(8020))
1934 self.APIBaseAPIBase = ":".join([self.APIHost,self.APIPort])
1935
1936 # Define custom redirect for current instance
1937 self.RedirectException(self.__pyx_api_delimn.join(["",str(2),str(PyXMake.__name__),"api","documentation"]))
1938
1939 # Mount static HTML files created by DoxyGen to created web application
1940 self.StaticFiles(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),"dev","documentation"]),self.__pyx_doc_path)
1941
1942 # Allow cross-site references
1943 self.APIAPI.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], allow_credentials=True)
1944
1945 # Add all API methods explicitly
1946 Backend.__init__(self, *args, **kwargs)
1947
1948if __name__ == '__main__':
1949 pass
Class instance to define PyXMake's web API instance.
__init__(self, *args, **kwargs)
Definition __init__.py:147
Abstract base class for all API objects.
run(self, Hostname=str(platform.node()), PortID=8020)
Definition __init__.py:130
StaticFiles(self, url, path, index="index.html", html=True)
Definition __init__.py:98
include(self, *args)
Definition __init__.py:115
mount(self, *args)
Definition __init__.py:109
__init__(self, *args, **kwargs)
Definition __init__.py:85
RedirectException(self, url)
Definition __init__.py:93
Class instance to define PyXMake's web API instance.
__init__(self, *args, **kwargs)
Definition __init__.py:1904
Abstract meta class for all data class objects.
Class to create 2to3 compatible pickling dictionary.
Class to create 2to3 compatible pickling dictionary.
Create a make object to define the building environment.
Definition Make.py:1
Module containing all relevant modules and scripts associated with the building process.
Definition __init__.py:1
Module containing basic functionalities defined for convenience.
Definition __init__.py:1