149 Initialization of PyXMake's Backend API.
158 Archive = Enum(
"ArchiveObjectKind",{
"zip":
"zip",
"tar":
"tar",
"gzip":
"gzip",
"lzma":
"lzma",
"bz2":
"bz2"})
160 FileManager = Enum(
"FileObjectKind",{
"full":
"all",
"local":
"private",
"shared":
"public"})
162 for key, value
in __attributes__.items(): setattr(self, key.replace(value,self.
APIObjectKind), getattr(self, key))
168 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"api",
"documentation"]), tags=[str(self.__pyx_guide)])
169 def api_self_guide():
171 Method pointing to the main documentation of the API.
173 return RedirectResponse(url=self.__pyx_api_delimn.join([
"",str(PyXMake.__name__),
"api",
"documentation",
""]))
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():
178 Method pointing to the user documentation of the underlying package.
180 response = str(self.__pyx_url_path)
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>
189 return HTMLResponse(html_content)
191 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"dev",
"documentation"]), tags=[str(self.__pyx_guide)])
194 Method pointing to the developer documentation of the underlying package.
196 return RedirectResponse(url=self.__pyx_api_delimn.join([
"",str(PyXMake.__name__),
"dev",
"documentation",
""]))
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(...)):
204 API for PyXMake to create an ABAQUS compatible Fortran library for Windows machines by using the Intel Fortran Compiler.
207 if os.path.exists(os.path.join(VTL.Scratch,ZIP.filename)):
208 os.remove(os.path.join(VTL.Scratch,ZIP.filename))
210 with Utility.TemporaryDirectory(VTL.Scratch),
Utility.UpdateZIP(ZIP.filename, ZIP.file, VTL.Scratch, update=
False):
212 files = Utility.FileWalk(Source, path=os.getcwd())
214 with Utility.FileOutput(
'result.log'):
215 VTL.abaqus(BuildID, files, source=os.getcwd(), scratch=os.getcwd(), output=os.getcwd(), verbosity=2)
217 return FileResponse(path=os.path.join(VTL.Scratch,ZIP.filename),filename=ZIP.filename)
219 @self.Router.patch(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"api",
"client",
"{kind}"]), tags=[str(self.__pyx_interface)],
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.
228 URI: str = Query(...,
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.
235 ClientID: Optional[str] = Query(
"python",
238 This is the generated client format. Defaults to python.
241 CLI: List[str] = Query([
"--skip-validate-spec"],
244 Additional command line arguments. Please refer to the official documentation of OpenAPI generator to get a glimpse of all available options.
247 ZIP: UploadFile = File(
None,
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.
256 API to connect to the OpenAPI client generator.
261 from pathlib
import Path
262 from packaging.version
import parse
267 canditate = str(next(tempfile._get_candidate_names()));
270 if not ZIP: ZIP = Utility.UpdateZIP.create(delimn.join([canditate,
"zip"]))
274 if os.path.exists(os.path.join(VTL.Scratch,ZIP.filename)): os.remove(os.path.join(VTL.Scratch,ZIP.filename))
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:])
283 image =
"docker.io/openapitools/openapi-generator-cli"
286 mount = [{
"ReadOnly":
False,
"Source":
"stmlab_scratch",
"Target":
"/mnt",
"Type":
"volume"}]
289 openapi_cli = [
"docker",
"run",
"--rm",
"-v",
"/mnt:%s" % root,image.split(posixpath.sep, 1)[-1]]
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)
301 url =
'{uri.scheme}://{uri.netloc}'.format(uri= urllib.parse.urlparse(response.url))
303 with open(URI,
'r')
as openfile: openapi_file = json.load(openfile)
305 servers = [{
'url': posixpath.sep.join([url,Utility.PathLeaf(x[
"url"])]).rstrip(posixpath.sep)}
for x
in openapi_file.get(
"servers",[{
"url":
""}])]
307 if parse(openapi_file.get(
"openapi")) >= parse(
"3.1.0"): openapi_file.update({
'openapi':
'3.0.3'})
309 openapi_file.update({
'servers':servers})
311 with open(URI,
'w')
as openfile: openfile.write(json.dumps(openapi_file))
315 if Utility.GetExecutable(
"java")
and Utility.GetOpenAPIGenerator():
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()];
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))]
325 command = openapi_cli + command
327 ind = command.index(
"generate") + 1; command[ind:ind] = CLI
330 data = {
"image":image,
"command":command[6:],
"keep":
False,
"template":{
"TaskTemplate":{
"ContainerSpec":{
"Mounts":mount}}}}
333 if not command[0]
in [
"docker"]: subprocess.check_call(command)
337 requests.post(str(request.url.scheme)+
"://%s/0/Lab/UI/Orchestra" % self.
APIBase,params={
"name":
"stm_openapi"}, json=data)
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})
345 if Utility.GetPlatform()
not in [
"windows"]: subprocess.check_call([
"sudo",
"chmod",
"-R",str(777),os.path.abspath(os.getcwd())])
348 return FileResponse(path=os.path.join(VTL.Scratch,ZIP.filename),filename=ZIP.filename)
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:
353 Dummy function for showing a quick response
357 return {
"return_code": string}
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:
362 Dummy function for showing a quick response
366 return {
"return_code": string}
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(...)):
371 API for PyXMake to create a static library for Windows machines by using the Intel Fortran Compiler.
374 if os.path.exists(os.path.join(VTL.Scratch,ZIP.filename)):
375 os.remove(os.path.join(VTL.Scratch,ZIP.filename))
377 with Utility.TemporaryDirectory(VTL.Scratch),
Utility.UpdateZIP(ZIP.filename, ZIP.file, VTL.Scratch, update=
False):
379 files = Utility.FileWalk(Source, path=os.getcwd())
381 os.makedirs(
"include",exist_ok=
True); os.makedirs(
"lib",exist_ok=
True)
383 with Utility.FileOutput(
'result.log'):
384 VTL.ifort(BuildID, files, source=os.getcwd(), scratch=os.getcwd(), make=os.getcwd(), verbosity=2)
386 return FileResponse(path=os.path.join(VTL.Scratch,ZIP.filename),filename=ZIP.filename)
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(...)):
391 API for PyXMake to create a Java compatible Fortran library for Windows machines by using the Intel Fortran Compiler.
394 if os.path.exists(os.path.join(VTL.Scratch,ZIP.filename)):
395 os.remove(os.path.join(VTL.Scratch,ZIP.filename))
397 with Utility.TemporaryDirectory(VTL.Scratch),
Utility.UpdateZIP(ZIP.filename, ZIP.file, VTL.Scratch, update=
False):
399 files = Utility.FileWalk(Source, path=os.getcwd())
401 with Utility.FileOutput(
'result.log'):
402 VTL.java(BuildID, files, source=os.getcwd(), scratch=os.getcwd(), output=os.getcwd(), verbosity=2)
404 return FileResponse(path=os.path.join(VTL.Scratch,ZIP.filename),filename=ZIP.filename)
406 @self.Router.patch(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"api",
"latex",
"{kind}"]), tags=[str(self.__pyx_interface)],
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.
411 def api_latex(kind: Archive,
413 ZIP: UploadFile = File(...,
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.
421 API for PyXMake to compile a Latex document remotely using Overleaf.
424 if os.path.exists(os.path.join(VTL.Scratch,ZIP.filename)): os.remove(os.path.join(VTL.Scratch,ZIP.filename))
426 with Utility.TemporaryDirectory(VTL.Scratch):
430 with Utility.FileOutput(
'result.log'): VTL.latex(BuildID, ZIP.filename, API=
"Overleaf", verbosity=2, keep=
False)
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)
435 return FileResponse(path=os.path.join(VTL.Scratch,ZIP.filename),filename=ZIP.filename)
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(...)):
440 API for PyXMake to create a Python compatible Fortran library for Windows machines by using the Intel Fortran Compiler.
443 if os.path.exists(os.path.join(VTL.Scratch,ZIP.filename)):
444 os.remove(os.path.join(VTL.Scratch,ZIP.filename))
446 with Utility.TemporaryDirectory(VTL.Scratch),
Utility.UpdateZIP(ZIP.filename, ZIP.file, VTL.Scratch, update=
False):
448 files = Utility.FileWalk(Source, path=os.getcwd())
450 with Utility.FileOutput(
'result.log'):
451 VTL.py2x(BuildID, files, source=os.getcwd(), scratch=os.getcwd(), output=os.getcwd(), verbosity=2)
453 return FileResponse(path=os.path.join(VTL.Scratch,ZIP.filename),filename=ZIP.filename)
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:
458 Dummy function for showing a long response time
461 return {
"return_code":str(time_sleep)}
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:
466 Dummy function for showing a long response time
469 return {
"return_code":str(time_sleep)}
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:
474 Dummy function for showing a long response time
477 return {
"return_code":str(time_sleep)}
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):
485 API to remotely access PyXMake methods.
488 import PyXMake
as pyx
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,)
498 @self.Router.head(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"api",
"user",
"{kind}"]), tags=[str(self.__pyx_professional)],
501 API for user info service. Verifies that a given token can be verified against a given service.
503 def api_get_user(request: Request,
504 kind: Enum(
"TokenObjectKind",{
"portainer":
"portainer",
"shepard":
"shepard",
"gitlab":
"gitlab",
"user":
"client"}),
505 Token: SecretStr = Query(...,
511 ClientID: str = Query(
None,
514 An user-defined client. Only meaningful when a non-default client is requested.
518 API to verify a given token. Returns information about the token holder if exists.
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)
527 elif kind.name
in [
"user"]
and not ClientID:
return JSONResponse(status_code=406, content=
"ClientID cannot be blank in non-default client mode")
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)
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),)
538 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"api",
"info",
"{kind}"]), tags=[str(self.__pyx_professional)],
541 API for user info service. Verifies that a given attribute can be found in a given category.
544 kind: Enum(
"UserObjectKind",{
"dlr-username":
"user",
"email":
"mail"}),
545 attribute: str = Query(...,
548 API to verify that a given user actual exists or is another service.
553 from PyCODAC.Tools.IOHandling
import Shepard
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",)
559 if result:
return JSONResponse(status_code=status_code,content=result[0],)
561 else:
return JSONResponse(status_code=status_code,content=
"There is no match for the given attribute in the category %s." % kind.value,)
563 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"api",
"generator",
"{kind}"]), tags=[str(self.__pyx_professional)], include_in_schema=
False,
566 API for generator service. Generates a random response depending on the given path parameter.
568 def api_generator(request: Request, kind: Enum(
"GeneratorObjectKind",{
"message":
"message"})):
570 Generates a random response in dependence of a given path parameter.
575 result =
""; gist =
"6440b706a97d2dd71574769517e7ed32"
578 with Utility.TemporaryDirectory(VTL.Scratch):
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)
582 with io.open(os.path.join(gist,
"loading_messages.js"),mode=
"r",encoding=
"utf-8")
as f: result = f.read()
584 result = result.replace(
"\n",
"").replace(
"export default ",
"").replace(
";",
"")
585 os.environ[
"pyx_message"] = result
587 try: git.rmtree(gist)
590 result = ast.literal_eval(os.getenv(
"pyx_message",result)); result = random.choice(result);
593 result =
"Service not available"; status_code = 500
595 return JSONResponse(status_code=status_code,content=result,)
597 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"api",
"database"]), tags=[str(self.__pyx_professional)], include_in_schema=
False,
600 API for internal database service. Use to collect information about all available internal database references.
604 API to fetch information about all internal databases.
607 result = {}; status_code = 500
609 from PyCODAC.Database
import Mongo
611 try: result = {
"DataContainer" : Mongo.base_table,
"DataTypes" : Mongo.base_reference }
614 try:
return (Enum(
"DataObjectKind",result[
"DataContainer"]),result[
"DataTypes"] )
616 if result: status_code = 200
618 return JSONResponse(status_code=status_code,content=result,)
620 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"api",
"license"]), tags=[str(self.__pyx_professional)],
623 API for FlexLM license manager services. Use to collect information about all supported license servers.
625 def api_license(request: Request):
627 API to fetch FlexLM license server information
630 result = {}; status_code = 500
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"]
643 return JSONResponse(status_code=status_code,content=result,)
645 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"api",
"flexlm"]), tags=[str(self.__pyx_professional)], include_in_schema=
False,
648 API for FlexLM license manager services. Returns the complete response as a JSON string.
650 def api_flexlm(license_query: Optional[str] = Query(
"27018@abaqussd1", description=
652 Qualified name or IP address and port number of the license server. The license server must be active and supporting FlexLM.
656 API to fetch FlexLM all license server information
660 from collections
import defaultdict
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",)
670 result = defaultdict(list)
673 with Utility.TemporaryDirectory():
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)
679 with Utility.FileOutput(tmp): Utility.SSHPopen(connect.ssh_client, ntpath.pathsep.join([command]),verbosity=2)
681 with open(tmp,
"r")
as f: content = f.read()
683 result = json.dumps(content); status_code = 200
685 except TimeoutError:
return JSONResponse(status_code=404,content=
"FlexLM service not reachable",)
688 return JSONResponse(status_code=status_code,content=result,)
690 @self.Router.post(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"api",
"flexlm",
"show"]), tags=[str(self.__pyx_professional)],
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.
696 def api_flexlm_show(request: Request,
697 license_query: Optional[str] = Query(
"27018@abaqussd1.intra.dlr.de",
700 Qualified name or IP address and port number of the license server. The license server must be active and supporting FlexLM.
703 license_attributes: List[str] = Body([
'abaqus',
'cae'],
706 List of attributes to be searched for in the FlexLM response. Defaults to ABAQUS CAE and solver token occupancy.
710 API to fetch FlexLM license occupancy information
712 from collections
import defaultdict
715 result = defaultdict(list)
718 license_request = defaultdict(
lambda:[1,0])
720 for x
in license_attributes: license_request[x]
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())
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]))
736 result = json.dumps(result); status_code = 200
739 return JSONResponse(status_code=status_code,content=result,)
741 @self.Router.post(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"api",
"flexlm",
"inspect"]), tags=[str(self.__pyx_professional)],
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.
746 def api_flexlm_inspect(request: Request,
747 license_query: Optional[str] = Query(
"27018@abaqussd1.intra.dlr.de",
750 Qualified name or IP address and port number of the license server. The license server must be active and supporting FlexLM.
753 license_attributes: List[str] = Body([
'abaqus',
'cae'],
756 List of attributes to be searched for in the FlexLM response. Defaults to ABAQUS CAE and solver token user information.
760 API to fetch FlexLM license holder information and number of licenses in use.
762 from collections
import defaultdict
765 result = defaultdict(list)
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())
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)])
781 result = json.dumps(result); status_code = 200
784 return JSONResponse(status_code=status_code,content=result,)
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)):
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.
793 import importlib, contextlib
796 try:
import PyCODAC.Core
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
803 filename =
'Default.zip';
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")
809 with Utility.TemporaryDirectory(VTL.Scratch),
Utility.UpdateZIP(filename, archive, VTL.Scratch, update=
False):
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([])
815 if not ModuleID
and default: mod = importlib.import_module(default)
816 else: mod = importlib.import_module(ModuleID)
821 from importlib
import reload
822 sys.path.insert(0,os.getcwd())
826 try: AllModules = jsonable_encoder(Py2X.inspect(mod))
827 except: AllModules = JSONResponse(status_code=404,content={
"message":
"No module named %s" % ModuleID
or default},)
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)):
833 Return the doc string of an existing symbol in a shared library.
834 Optionally, upload a compiled Python extension library.
837 import importlib, contextlib
840 try:
import PyCODAC.Core
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
847 filename =
'Default.zip';
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")
853 with Utility.TemporaryDirectory(VTL.Scratch),
Utility.UpdateZIP(filename, archive, VTL.Scratch, update=
False):
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([])
860 if not ModuleID
and default: mod = importlib.import_module(default)
861 else: mod = importlib.import_module(ModuleID)
866 from importlib
import reload
867 sys.path.insert(0,os.getcwd())
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])},)
875 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"api",
"gitlab",
"projects"]), tags=[str(self.__pyx_professional)], include_in_schema=
False,
878 API for GitLab group service. List all projects inside a given group.
880 def api_gitlab_projects(
881 Token: SecretStr =Query(...,
884 GitLab X-API-Token valid for the group.
887 GroupID: Optional[str] = Query(
"6371",
890 The group identifier. Defaults to group STMLab.
894 API to fetch information about all projects within a group.
897 from PyXMake.VTL
import gitlab
899 url, header = gitlab.datacheck(token=Token.get_secret_value())
901 r = requests.get(posixpath.join(url,
"groups",str(GroupID),
"projects"), headers= header)
903 return JSONResponse(json.dumps({x[
"name"]:x[
"id"]
for x
in r.json()}))
905 @self.Router.post(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"api",
"gitlab",
"groups"]), tags=[str(self.__pyx_professional)], include_in_schema=
False,
908 API for GitLab. List all groups avoiding decoding error. A dictionary for filtering the results can be provided.
910 def api_gitlab_groups(
911 Token: SecretStr =Query(...,
914 GitLab X-API-Token valid for the instance.
917 Filter: Optional[dict] = Body({
"name":
"STMLab"},
920 Filtering results for key, value pairs. Defaults to filtering for group STMLab.
924 API to fetch information about all groups within an instance.
928 from PyXMake.VTL
import gitlab
930 url, header = gitlab.datacheck(token=Token.get_secret_value())
932 status_code = 500; result = {}
934 r = requests.get(posixpath.join(url,
"groups?per_page=100"), headers=header)
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
940 if Filter: result = [x
for x
in result
if all([x[key]
in [value]
for key, value
in Filter.items()])]
943 return JSONResponse(status_code=status_code,content=result,)
945 @self.Router.post(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"api",
"gitlab",
"packages"]), tags=[str(self.__pyx_professional)], include_in_schema=
False,
948 API for GitLab group service. List all packages available within a given group.
950 def api_gitlab_packages(
951 Token: SecretStr =Query(...,
954 GitLab X-API-Token valid for the group.
957 GroupID: Optional[str] = Query(
"6371",
960 The group identifier. Defaults to group STMLab.
963 Filter: Optional[dict] = Body({
"package_type":
"generic",
"name":
"STMLab"},
966 Filtering results for key, value pairs. Defaults to filtering for the installer of STMLab.
970 API to fetch information about all packages with a group sorted by their version.
974 from PyXMake.VTL
import gitlab
976 url, header = gitlab.datacheck(token=Token.get_secret_value())
978 status_code = 500; result = {}
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;
986 return JSONResponse(status_code=status_code,content=result,)
988 @self.Router.put(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"api",
"gitlab",
"job"]), tags=[str(self.__pyx_professional)], include_in_schema=
False,
991 API for GitLab job service. Run a given project pipeline job with non-default environment variables.
994 Token: SecretStr = Query(...,
997 GitLab X-API-Token valid for the project.
1000 ProjectID: Optional[str] = Query(str(12702),
1003 The project identifier. Defaults to PyXMake's CI pipeline.
1006 Job: Optional[str] = Query(
"pyx_core_docs",
1009 A job name. Defaults to a pipeline job for documenting PyXMake.
1013 API to run a GitLab CI job with non-default environment variables.
1016 from PyXMake.VTL
import gitlab
1018 return JSONResponse(json.dumps(gitlab.pipeline(Token.get_secret_value(), ProjectID, job_name=Job)))
1020 @self.Router.put(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"api",
"gitlab",
"pipeline"]), tags=[str(self.__pyx_professional)], include_in_schema=
False,
1023 API for GitLab pipeline service. Run a given project pipeline with non-default environment variables.
1025 def api_gitlab_pipeline(
1026 Token: SecretStr = Query(...,
1029 GitLab X-API-Token valid for the project.
1032 ProjectID: Optional[str] = Query(str(12702),
1035 The project identifier. Defaults to PyXMake's CI pipeline.
1039 API to run a GitLab CI pipeline.
1042 from PyXMake.VTL
import gitlab
1044 return JSONResponse(json.dumps(gitlab.pipeline(Token.get_secret_value(), ProjectID)))
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},},
1052 API for GitLab pipeline service. Interact with a given group-level variable.
1054 async def api_gitlab_variable(
1056 kind: Enum(
"VariableObjectKind",{
"groups":
"group",
"projects":
"project"}),
1057 Token: SecretStr = Query(...,
1060 GitLab X-API-Token valid for the call.
1063 ID: str = Query(str(6371),
1066 The project or group identifier. Defaults to group STMLab.
1070 API shim to interact with GitLab variables on both project and group level.
1073 from PyXMake.VTL
import gitlab
1075 url, header = gitlab.datacheck(token=Token.get_secret_value())
1077 method = str(request.method).lower()
1079 body=b
""; headers={
"Content-Type":
"application/json"}; result = {}; status_code = 500;
1082 body = await request.body()
1084 if body: body = json.loads(body)
1085 url = posixpath.join(url,str(kind.name), ID,
"variables");
1087 if isinstance(body,dict)
and not method
in [
"post"]: url = posixpath.join(url,body.get(
"key",
""));
1089 r = requests.request(method, url, params=request.query_params, data=body, headers=header)
1091 status_code = r.status_code; result = r.content; headers.update(r.headers)
1092 try: headers.pop(
"Content-Encoding")
1096 return Response(status_code=status_code, content=result, headers=headers)
1098 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"api",
"gitlab",
"runner"]), tags=[str(self.__pyx_professional)], include_in_schema=
False,
1101 API for GitLab service. Creates a GitLab runner dedicated for the given project registration token on FA-Services.
1103 def api_gitlab_runner(
1104 Registration: str = Query(...,
1107 GitLab registration token (mandatory).
1110 Token: SecretStr = Query(
"",
1113 Portainer X-API-Token.
1117 API to create a GitLab runner dedicated for the given project registration token.
1122 is_service =
True; with_tmpfs =
True; with_persistent =
True
1123 tmpfs = []; mounts = []; result = {}; status_code = 500;
1125 from PyXMake.VTL
import portainer
1126 from PyCODAC.API.Remote
import Server
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))
1132 url, header = portainer.main(str(getattr(Token,
"get_secret_value",str)())
or Server.auth[
'X-API-KEY'], datacheck=
True);
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})
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"}])
1141 if mounts
and (with_tmpfs
and not with_persistent): tmpfs = [umlshm, persistent]
1142 elif mounts
and (with_tmpfs
and with_persistent): tmpfs = [umlshm]
1145 data = {
"Name": x,
"DriverOpts": {
"device":
"tmpfs",
"o":
"rw,nosuid,nodev,exec",
"type":
"tmpfs"}}
1147 r = requests.post(posixpath.join(url,
"endpoints",node,
"docker",
"volumes",
"create"),
1148 json=data, headers=header); r.raise_for_status()
1150 if with_persistent
and is_service:
1152 r = requests.post(posixpath.join(url,
"endpoints",node,
"docker",
"volumes",
"create"),
1153 json={
"Name":persistent}, headers=header); r.raise_for_status()
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}}}}
1164 r = requests.post(posixpath.join(url,
"endpoints",node,
"docker",
"services",
"create"), json=data, headers=header);
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()
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
1177 except Exception:
pass
1179 return JSONResponse(status_code=status_code,content=result,)
1181 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"api",
"gitlab",
"user"]), tags=[str(self.__pyx_professional)], include_in_schema=
False,
1184 API for GitLab service. Returns the current user for a given API token.
1186 def api_gitlab_user(
1187 Token: SecretStr = Query(...,
1194 API to get the current active user from an API Token.
1197 from PyXMake.VTL
import gitlab
1199 url, header = gitlab.main(token=Token.get_secret_value(), datacheck=
True)
1201 result = {}; status_code = 500;
1203 r = requests.get(posixpath.join(url,
"user"), headers= header); result = r.json()
1204 if result: status_code = r.status_code
1207 return JSONResponse(status_code=status_code,content=result,)
1209 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"api",
"portainer",
"user"]), tags=[str(self.__pyx_professional)], include_in_schema=
False,
1212 API for Portainer service. Returns the current user for a given API token.
1214 def api_portainer_user(
1215 Token: SecretStr = Query(...,
1218 Portainer X-API-Token.
1222 API to get the current active user from an API Token.
1225 status_code = 500; result = []
1227 from PyXMake.VTL
import portainer
1230 result = portainer.main(Token.get_secret_value());
1232 if isinstance(result, list):
1233 status_code = 200; result = result[0]
1235 else: status_code = 401
1238 return JSONResponse(status_code=status_code,content=result,)
1240 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"api",
"shepard",
"user"]), tags=[str(self.__pyx_professional)], include_in_schema=
False,
1243 API for Shepard service. Returns the current user for a given API token.
1245 def api_shepard_user(
1246 Token: SecretStr = Query(...,
1249 Shepard X-API-Token.
1253 API to get the current active user from an API Token.
1256 status_code = 500; result = {}
1259 from PyCODAC.Tools.IOHandling
import Shepard
1260 except:
return JSONResponse(status_code=404,content=
"Service not available",)
1263 result = Shepard.GetUser(header= {
'X-API-KEY':Token.get_secret_value()}, full=
True);
1265 if result: status_code = 200;
1268 return JSONResponse(status_code=status_code,content=result,)
1270 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"api",
"shepard",
"{kind}"]), tags=[str(self.__pyx_professional)], include_in_schema=
False,
1273 API for PyXMake. Browse through all publicly available Shepard identifiers present within the scope of this API.
1275 def api_shepard_show(
1276 kind: api_database()[0],
1277 ID: Optional[str] = Query(
None,
1280 An user-defined container ID.
1283 Token: SecretStr = Query(
None,
1286 Shepard X-API-Token.
1290 API to connect to all available data identifiers using PyXMake.
1293 result = []; status_code = 500; delimn =
"_";
1294 DB = ID
or str(kind.name); options = api_database()[-1]
1296 from PyCODAC.Tools.IOHandling
import Shepard
1298 try: Shepard.SetHeader( {
'X-API-KEY':Token.get_secret_value()} )
1300 finally: Shepard.GetContainerID(DB)
1301 except:
return JSONResponse(status_code=404,content=
"Service not available",)
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])
1308 except ConnectionRefusedError:
1310 if options[str(kind.value)]
in [
"timeseries"]: status_code = 200
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)]),)
1316 result = list(Utility.ArbitraryFlattening(result))
1317 if result: status_code = 200
1319 return JSONResponse(status_code=status_code, content=result,)
1321 @self.Router.delete(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"api",
"shepard",
"{kind}"]), tags=[str(self.__pyx_professional)], include_in_schema=
False,
1324 API for PyXMake. Delete an available data object identifier present within the scope of this API.
1326 def api_shepard_delete(
1327 kind: api_database()[0],
1328 OID: Optional[str] = Query(...,
1331 The object identifier of the data.
1334 ID: Optional[str] = Query(
None,
1337 An user-defined container ID.
1340 Token: SecretStr = Query(
None,
1343 Shepard X-API-Token.
1347 API to delete any available data identifier using PyXMake.
1350 result = []; status_code = 500; DB = ID
or str(kind.name); options = api_database()[-1]
1352 from PyCODAC.Tools.IOHandling
import Shepard
1354 try: Shepard.SetHeader( {
'X-API-KEY':Token.get_secret_value()} )
1356 finally: Shepard.GetContainerID(DB)
1357 except:
return JSONResponse(status_code=404,content=
"Service not available",)
1360 Shepard.DeleteContent(DB, OID, query=options[str(kind.value)], latest=
False)
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"
1367 return JSONResponse(status_code=status_code, content=result)
1369 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"api",
"overleaf",
"session"]), tags=[str(self.__pyx_professional)], include_in_schema=
False,
1372 API for Overleaf service. Creates a valid API session token for Overleaf to automate tasks.
1374 def api_overleaf_session(
1375 Token: SecretStr = Query(...,
1378 Credentials given as Overleaf X-API-Token with its value set to the result of:
1379 "echo -n "<'YourUserName'>:<'YourPasswort'>" | base64"
1382 base_url: Optional[str] = Query(
"", include_in_schema=
False,
1385 Base URL of the Overleaf instance in question. Defaults to Overleaf running on FA-Services.
1388 Creates a valid API session token for Overleaf.
1391 status_code = 500; 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);
1398 return JSONResponse(status_code=status_code,content=result,)
1400 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"api",
"overleaf",
"show"]), tags=[str(self.__pyx_professional)], include_in_schema=
False,
1403 API for Overleaf service. Compile an Overleaf project remotely and return all files.
1405 def api_overleaf_show(
1406 Token: SecretStr = Query(...,
1409 Credentials given as Overleaf X-API-Token with its value set to the result of:
1410 "echo -n "<'YourUserName'>:<'YourPasswort'>" | base64"
1413 ProjectID: str = Query(..., description=
1415 An Overleaf Project ID.
1418 base_url: Optional[str] = Query(
"", include_in_schema=
False, description=
1420 Base URL of the Overleaf instance in question. Defaults to Overleaf running on FA-Services.
1423 Compile an Overleaf project remotely and return all files.
1426 status_code = 500; result = {}
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 ;
1435 return JSONResponse(status_code=status_code,content=result,)
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,
1440 API for Overleaf service. Compile an Overleaf project remotely and return all files.
1442 def api_overleaf_download(
1443 kind: Enum(
"DataObjectKind",{
"zip":
"zip",
"pdf":
"pdf"}),
1444 Token: SecretStr = Query(...,
1447 Credentials given as Overleaf X-API-Token with its value set to the result of:
1448 "echo -n "<'YourUserName'>:<'YourPasswort'>" | base64"
1451 ProjectID: str = Query(..., description=
1453 An Overleaf Project ID.
1456 base_url: Optional[str] = Query(
"", include_in_schema=
False, description=
1458 Base URL of the Overleaf instance in question. Defaults to Overleaf running on FA-Services.
1461 Compile an Overleaf project remotely and return all files.
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",)
1472 return FileResponse(path=FilePath,filename=Utility.PathLeaf(FilePath))
1474 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"api",
"time"]), tags=[str(self.__pyx_professional)], include_in_schema=
False,
1477 API for time service. Get the current system time of this API in ISO 8601 format (UTC).
1479 def api_time_show(request: Request):
1481 Get the current system time of this API in ISO 8601 format (UTC).
1484 timestamp =
"Unknown system time"
1487 timestamp = str(datetime.datetime.now().strftime(
"%Y-%m-%dT%H:%M:%SZ"))
1489 except: status_code = 500
1491 return JSONResponse(status_code=status_code,content=timestamp,)
1493 @self.Router.post(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"api",
"time"]), tags=[str(self.__pyx_professional)], include_in_schema=
False,
1496 API for time service. Return the given time from any time format in UTC format (nanoseconds).
1498 def api_time_inspect(request: Request,
1499 Time: str = Query(...,
1502 Arbitrary valid time string. The string is parsed into an internal date object.
1506 Return the given time from any time format in UTC format (nanoseconds).
1508 from dateutil.parser
import parse
1510 timestamp =
"Unknown input time"
1513 timestamp = int(parse(Time).replace(tzinfo=datetime.timezone.utc).timestamp() * 1e9)
1515 except: status_code = 500
1517 return JSONResponse(status_code=status_code,content=timestamp,)
1519 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"api",
"secret"]), tags=[str(self.__pyx_professional)],
1522 API for PyXMake. Show a list of all available secrets for a given token.
1524 def api_secret_show(request: Request,
1525 Token: SecretStr = Query(...,
1528 Portainer X-API-Token. The token is stored in your personal user space.
1532 API to get the current active user from an API Token.
1535 from PyXMake.VTL
import portainer
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",)
1540 with Utility.TemporaryDirectory():
1542 url, header = portainer.main(Token.get_secret_value(), datacheck=
True);
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)
1547 return JSONResponse(status_code=r.status_code,content=r.json(),)
1549 @self.Router.post(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"api",
"secret"]), tags=[str(self.__pyx_professional)],
1552 API for PyXMake. Create a new secret for the given token.
1554 def api_secret_upload(request: Request,
1555 payload: UploadFile = File(...,
1561 Token: SecretStr = Query(...,
1564 Portainer X-API-Token. The token is stored in your personal user space.
1567 SecretID: str = Query(
None,
1570 An optional name for the secret. Defaults to the name of the given file.
1574 API to get the current active user from an API Token.
1577 from PyXMake.VTL
import portainer
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",)
1582 with Utility.TemporaryDirectory():
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()
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);
1590 body = {
'Data': content,
'Name':SecretID
or payload.filename,
'Labels':
None } ;
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)
1596 return JSONResponse(status_code=r.status_code,content=r.json(),)
1598 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"api",
"file",
"{kind}"]), tags=[str(self.__pyx_professional)],
1601 API for PyXMake. Browse through all available file identifiers present within the scope of this API.
1603 def api_file_show(request: Request, kind: FileManager):
1605 API to connect to all available file identifiers using PyXMake.
1608 result = []; status_code = 500;
1610 from PyCODAC.VTL
import Fetch
1611 except:
return JSONResponse(status_code=404,content=
"Service not available",)
1613 rd = random.Random();
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()
1620 if kind.value
in [
"all",
"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])
1626 for i,x
in enumerate(private):
1628 rd.seed(int(i)); x[next(iter(x))].
update({
"id": str(uuid.UUID(int=rd.getrandbits(128), version=4))})
1632 result = list(Utility.ArbitraryFlattening(result))
1633 if result: status_code = 200
1635 return JSONResponse(status_code=status_code, content=result,)
1637 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"api",
"file"]), tags=[str(self.__pyx_professional)],
1640 API for PyXMake. Download a file from an available data source by its id.
1642 def api_file_download(request: Request, FileID: str = Query(...)):
1644 API to upload a given file using PyXMake.
1647 FileDB = [data.name
for data
in api_database()[0]
if data.value == Utility.PathLeaf(str(request.url).split(
"?")[0])][0]
1649 from PyCODAC.VTL
import Fetch
1650 from PyCODAC.Tools.IOHandling
import Shepard
1651 Shepard.GetContainerID(FileDB)
1652 except:
return JSONResponse(status_code=404,content=
"Service not available",)
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
1659 except:
return JSONResponse(status_code=500,content=
"Internal database error: There are multiple files sharing the same ID.",)
1661 if not FileID
in ListofIDs:
return JSONResponse(status_code=404,content=
"File not found",)
1663 with Utility.TemporaryDirectory():
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]
1668 if FileDB.split(
"_")[-1].lower()
in [data[1]]: FilePath = Shepard.DownloadFile(FileDB, FileID, latest=
False)[0]
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.",)
1673 return FileResponse(path=FilePath,filename=Utility.PathLeaf(FilePath))
1675 @self.Router.post(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"api",
"file",
"{kind}"]), tags=[str(self.__pyx_professional)],
1678 API for PyXMake. Connects an user with an available file browser and uploads the given data.
1680 def api_file_upload(request: Request,
1681 kind: Enum(
"ScopeObjectKind",{
"local":
"private",
"shared":
"public"}),
1682 payload: UploadFile = File(...,
1689 API to upload a given file using PyXMake.
1694 FileDB = [data.name
for data
in api_database()[0]
if data.value ==Utility.PathLeaf(posixpath.dirname(str(request.url)))][0]
1696 if str(kind.value)
in [
"public"]:
1697 from PyCODAC.Tools.IOHandling
import Shepard
1698 Shepard.GetContainerID(FileDB)
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",)
1703 with Utility.TemporaryDirectory():
1705 from PyCODAC.Database
import Mongo
1707 with open(payload.filename,
'wb+')
as f: f.write(payload.file.read())
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)
1713 result = Shepard.GetContent(FileDB)[payload.filename]
1714 elif str(kind.value)
in [
"private"]:
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')]
1720 else:
raise NotImplementedError
1721 if result: status_code = 200
1723 return JSONResponse(status_code=status_code, content=result,)
1725 @self.Router.patch(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"api",
"file",
"format",
"{kind}"]), tags=[str(self.__pyx_professional)],
1728 API for PyXMake. Connects an user with various file formatting options present within the scope of this API.
1730 def api_file_converter(
1732 OutputID: Enum(
"FormatObjectKind",{
"xls":
"xls"}),
1733 SourceID: Optional[str] = Query(
"Settings.xlsx",
1736 The name of the main file present in the supplied ZIP archive. The requested output format is given by OutputID.
1738 ZIP: UploadFile = File(...,
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.
1747 API to connect to various file formatting options using PyXMake.
1750 if os.path.exists(os.path.join(VTL.Scratch,ZIP.filename)): os.remove(os.path.join(VTL.Scratch,ZIP.filename))
1753 with Utility.TemporaryDirectory(VTL.Scratch),
Utility.UpdateZIP(ZIP.filename, ZIP.file, VTL.Scratch, update=
True):
1755 with Utility.FileOutput(
'result.log'):
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(),) )
1763 return FileResponse(path=os.path.join(VTL.Scratch,ZIP.filename),filename=ZIP.filename)
1765 @self.Router.get(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"api",
"image",
"{kind}"]), tags=[str(self.__pyx_professional)],
1768 API for PyXMake. Obtain information about an image.
1770 def api_image_show(kind: Enum(
"ImageObjectKind",{
"public":
"public",
"user":
"user"}),
1772 image: str = Query(...,
1775 Fully qualified name of the image.
1779 API to obtain image information
1785 if kind.name
in [
"public"]
and not image.startswith(
"docker.io"): search = posixpath.join(
"docker.io",str(image))
1787 elif kind.name
in [
"user"]: search = str(image)
1788 else:
raise NotImplementedError
1790 base_url = str(request.url.scheme)+
"://%s/api/portainer" % self.
APIBase
1792 r = requests.get(posixpath.join(base_url,
"endpoints")); endpoint = [str(x[
"Id"])
for x
in r.json()][-1]
1794 requests.post(posixpath.join(base_url,
"endpoints",endpoint,
"docker",
"images",
"create"), params={
"fromImage":search})
1796 result.update(requests.get(posixpath.join(base_url,
"endpoints",endpoint,
"docker",
"images",image,
"json")).json())
1798 if result:
return JSONResponse(status_code=200, content=result)
1800 except:
return JSONResponse(status_code=404, content=
"Could not determine info for image %s. Image probably not in scope." % image)
1802 @self.Router.post(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"api",
"encoding",
"{kind}"]), tags=[str(self.__pyx_professional)],
1805 API for PyXMake. Encode a given secret string.
1807 def api_encode_message(kind: Enum(
"EncodingObjectKind",{
"base64":
"base64"}),
1808 message: SecretStr = Query(...,
1811 Message (string) to be encoded
1815 API to encode a message
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.")
1824 @self.Router.put(self.__pyx_api_delimn.join(["",str(PyXMake.__name__),
"api",
"notification",
"{kind}"]), tags=[str(self.__pyx_professional)],
1827 API for PyXMake. Connect with a notification service to send arbitrary mail or Mattermost notifications.
1829 def api_notification(
1830 kind: Enum(
"NotificationObjectKind",{
"mail":
"mail",
"mattermoost":
"mattermost"}),
1831 recipient: str = Query(..., description=
1833 Qualified mail address or mattermost channel.
1836 notification: dict = Body({
"subject":
"MyFancyHeader",
"content":
"MyFancyMessage"},
1839 A dictionary containing the subject line and the content.
1843 API for PyXMake. Connect with a notification service.
1847 from collections
import defaultdict
1852 default_user =
"f_testpa"; default_host =
"cara.dlr.de"
1853 secrets = os.path.join(Utility.AsDrive(
"keys"),default_host)
1855 if kind.value
in [
"mattermost"]:
1858 payload={
"text": notification.pop(
"text",
"")
or "\n".join(
1859 [notification.pop(
"subject"),notification.pop(
"content")]),
"username":notification.pop(
"username",str(kind.value).title())}
1861 payload.update(notification)
1862 requests.post(recipient, json=payload)
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",)
1871 with Utility.TemporaryDirectory():
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)
1877 with Utility.FileOutput(tmp): Utility.SSHPopen(connect.ssh_client, ntpath.pathsep.join([command]),verbosity=2)
1879 with open(tmp,
"r")
as f: content = f.read()
1881 _ = json.dumps(content); status_code = 200
1883 except TimeoutError:
return JSONResponse(status_code=404,content=
"%s service not available" % str(kind.value).title(),)
1886 return JSONResponse(status_code=status_code,content=
"Success",)