Auto generate code for resource objects

.pre-commit-config.yaml

@@ -4,16 +4,19 @@ repos:
   rev: 22.3.0
   hooks:
   - id: black
+    exclude: ^(ansys/rep/client/jms/resource/|ansys/rep/client/auth/resource/)
 
 - repo: https://github.com/pycqa/isort
   rev: 5.10.1
   hooks:
   - id: isort
+    exclude: ^(ansys/rep/client/jms/resource/|ansys/rep/client/auth/resource/)
 
 - repo: https://gitlab.com/PyCQA/flake8
   rev: 4.0.1
   hooks:
   - id: flake8
+    exclude: ^(ansys/rep/client/jms/resource/|ansys/rep/client/auth/resource/)
 
 - repo: https://github.com/codespell-project/codespell
   rev: v2.1.0

ansys/rep/client/auth/api/auth_api.py

@@ -6,7 +6,9 @@
 # Author(s): F.Negri, O.Koenig
 # ----------------------------------------------------------
 
-from ..resource.user import create_user, delete_user, get_users, update_user
+from keycloak import KeycloakAdmin
+
+from ..schema.user import UserSchema
 
 
 class AuthApi:
@@ -71,3 +73,82 @@ def delete_user(self, user):
             user (:class:`ansys.rep.client.auth.User`): A User object. Defaults to None.
         """
         return delete_user(self.client, user)
+
+
+def _admin_client(client):
+
+    custom_headers = {
+        "Authorization": "Bearer " + client.access_token,
+        "Content-Type": "application/json",
+    }
+    keycloak_admin = KeycloakAdmin(
+        server_url=client.auth_api_url,
+        username=None,
+        password=None,
+        realm_name=client.realm,
+        client_id=client.client_id,
+        verify=False,
+        custom_headers=custom_headers,
+    )
+    return keycloak_admin
+
+
+def get_users(client, as_objects=True):
+    admin = _admin_client(client)
+    data = admin.get_users({})
+    for d in data:
+        uid = d["id"]
+        groups = admin.get_user_groups(uid)
+        d["groups"] = [g["name"] for g in groups]
+        realm_roles = admin.get_realm_roles_of_user(uid)
+        d["realm_roles"] = [r["name"] for r in realm_roles]
+        d["is_admin"] = d  # Force admin check
+
+    schema = UserSchema(many=True)
+    users = schema.load(data)
+    return users
+
+
+def create_user(client, user, as_objects=True):
+    schema = UserSchema(many=False)
+    data = schema.dump(user)
+
+    pwd = data.pop("password", None)
+    if pwd is not None:
+        data["credentials"] = [
+            {
+                "type": "password",
+                "value": pwd,
+            }
+        ]
+    data["enabled"] = True
+
+    admin = _admin_client(client)
+    uid = admin.create_user(data)
+    data = admin.get_user(uid)
+    user = schema.load(data)
+    return user
+
+
+def update_user(client, user, as_objects=True):
+    schema = UserSchema(many=False)
+    data = schema.dump(user)
+
+    pwd = data.pop("password", None)
+    if pwd is not None:
+        data["credentials"] = [
+            {
+                "type": "password",
+                "value": pwd,
+            }
+        ]
+
+    admin = _admin_client(client)
+    data = admin.update_user(user.id, data)
+    user = schema.load(data)
+    return user
+
+
+def delete_user(client, user):
+    admin = _admin_client(client)
+    admin.delete_user(user.id)

ansys/rep/client/auth/resource/user.py

@@ -1,123 +1,49 @@
-# ----------------------------------------------------------
-# Copyright (C) 2019 by
-# ANSYS Switzerland GmbH
-# www.ansys.com
-#
-# Author(s): F.Negri
-# ----------------------------------------------------------
-import logging
-
-from keycloak import KeycloakAdmin
+# autogenerated code based on UserSchema
 
+from marshmallow.utils import missing
 from ansys.rep.client.jms.resource.base import Object
-
 from ..schema.user import UserSchema
 
-log = logging.getLogger(__name__)
-
-
 class User(Object):
-    """User resource
-
-    Args:
-        **kwargs: Arbitrary keyword arguments, see the User schema below.
-
-    Example:
-
-        >>> new_user = User(username='test_user', password='dummy',
-        >>>         email='test_user@test.com', fullname='Test User',
-        >>>         is_admin=False)
-
-    The User schema has the following fields:
-
-    .. jsonschema:: schemas/User.json
+    """User resource.
+
+    Parameters
+    ----------
+    id : str
+        Unique user ID, generated internally by the server on creation.
+    username : str
+        Username.
+    password : str
+        Password.
+    first_name : str, optional
+        First name
+    last_name : str, optional
+        Last name
+    email : str, optional
+        E-mail address (optional).
+    groups : str
+        Groups the user belongs to
+    realm_roles : str
+        Realm roles assigned to the user
+    is_admin        Whether the user has admin rights or not.
 
     """
 
     class Meta:
         schema = UserSchema
+        rest_name = "None"
 
     def __init__(self, **kwargs):
-        super(User, self).__init__(**kwargs)
-
+        self.id = missing
+        self.username = missing
+        self.password = missing
+        self.first_name = missing
+        self.last_name = missing
+        self.email = missing
+        self.groups = missing
+        self.realm_roles = missing
+        self.is_admin = missing
+
+        super().__init__(**kwargs)
 
 UserSchema.Meta.object_class = User
-
-
-def _admin_client(client):
-
-    custom_headers = {
-        "Authorization": "Bearer " + client.access_token,
-        "Content-Type": "application/json",
-    }
-    keycloak_admin = KeycloakAdmin(
-        server_url=client.auth_api_url,
-        username=None,
-        password=None,
-        realm_name=client.realm,
-        client_id=client.client_id,
-        verify=False,
-        custom_headers=custom_headers,
-    )
-    return keycloak_admin
-
-
-def get_users(client, as_objects=True):
-    admin = _admin_client(client)
-    data = admin.get_users({})
-    for d in data:
-        uid = d["id"]
-        groups = admin.get_user_groups(uid)
-        d["groups"] = [g["name"] for g in groups]
-        realm_roles = admin.get_realm_roles_of_user(uid)
-        d["realm_roles"] = [r["name"] for r in realm_roles]
-        d["is_admin"] = d  # Force admin check
-
-    schema = UserSchema(many=True)
-    users = schema.load(data)
-    return users
-
-
-def create_user(client, user, as_objects=True):
-    schema = UserSchema(many=False)
-    data = schema.dump(user)
-
-    pwd = data.pop("password", None)
-    if pwd is not None:
-        data["credentials"] = [
-            {
-                "type": "password",
-                "value": pwd,
-            }
-        ]
-    data["enabled"] = True
-
-    admin = _admin_client(client)
-    uid = admin.create_user(data)
-    data = admin.get_user(uid)
-    user = schema.load(data)
-    return user
-
-
-def update_user(client, user, as_objects=True):
-    schema = UserSchema(many=False)
-    data = schema.dump(user)
-
-    pwd = data.pop("password", None)
-    if pwd is not None:
-        data["credentials"] = [
-            {
-                "type": "password",
-                "value": pwd,
-            }
-        ]
-
-    admin = _admin_client(client)
-    data = admin.update_user(user.id, data)
-    user = schema.load(data)
-    return user
-
-
-def delete_user(client, user):
-    admin = _admin_client(client)
-    admin.delete_user(user.id)

ansys/rep/client/jms/api/project_api.py

@@ -1,20 +1,25 @@
+import json
 import logging
 import os
 from pathlib import Path
 from typing import Callable, List
 
+from cachetools import TTLCache, cached
+from marshmallow.utils import missing
+import requests
+
 from ansys.rep.client.exceptions import ClientError, REPError
 
 from ..resource.algorithm import Algorithm
 from ..resource.base import Object
 from ..resource.file import File
-from ..resource.job import Job, copy_jobs, sync_jobs
+from ..resource.job import Job, JobSchema
 from ..resource.job_definition import JobDefinition
 from ..resource.license_context import LicenseContext
 from ..resource.parameter_definition import ParameterDefinition
 from ..resource.parameter_mapping import ParameterMapping
-from ..resource.project import get_fs_url
-from ..resource.project_permission import ProjectPermission, update_permissions
+from ..resource.project import Project
+from ..resource.project_permission import ProjectPermission, ProjectPermissionSchema
 from ..resource.selection import JobSelection
 from ..resource.task import Task
 from ..resource.task_definition import TaskDefinition
@@ -187,10 +192,10 @@ def delete_job_definitions(self, job_definitions):
 
     ################################################################
     # Jobs
-    def get_jobs(self, as_objects=True, **query_params):
+    def get_jobs(self, as_objects=True, **query_params) -> List[Job]:
         return self._get_objects(Job, as_objects=as_objects, **query_params)
 
-    def create_jobs(self, jobs, as_objects=True):
+    def create_jobs(self, jobs, as_objects=True) -> List[Job]:
         """Create new jobs
 
         Args:
@@ -213,7 +218,7 @@ def copy_jobs(self, jobs, as_objects=True, **query_params):
         """
         return copy_jobs(self, jobs, as_objects=as_objects, **query_params)
 
-    def update_jobs(self, jobs, as_objects=True):
+    def update_jobs(self, jobs, as_objects=True) -> List[Job]:
         """Update existing jobs
 
         Args:
@@ -489,3 +494,62 @@ def archive_project(project_api: ProjectApi, target_path, include_job_files=True
 
     log.info(f"Done saving project archive to disk")
     return file_path
+
+
+def copy_jobs(project_api: ProjectApi, jobs: List[Job], as_objects=True, **query_params):
+    """Create new jobs by copying existing ones"""
+
+    url = f"{project_api.url}/jobs"
+
+    query_params.setdefault("fields", "all")
+
+    json_data = json.dumps({"source_ids": [obj.id for obj in jobs]})
+    r = project_api.client.session.post(f"{url}", data=json_data, params=query_params)
+
+    data = r.json()["jobs"]
+    if not as_objects:
+        return data
+
+    return JobSchema(many=True).load(data)
+
+
+def sync_jobs(project_api: ProjectApi, jobs: List[Job]):
+
+    url = f"{project_api.url}/jobs:sync"
+    json_data = json.dumps({"job_ids": [obj.id for obj in jobs]})
+    r = project_api.client.session.put(f"{url}", data=json_data)
+
+
+def update_permissions(client, project_api_url, permissions):
+
+    if not permissions:
+        return
+
+    url = f"{project_api_url}/permissions"
+
+    schema = ProjectPermissionSchema(many=True)
+    serialized_data = schema.dump(permissions)
+    json_data = json.dumps({"permissions": serialized_data})
+    r = client.session.put(f"{url}", data=json_data)
+
+
+@cached(cache=TTLCache(1024, 60), key=lambda project: project.id)
+def get_fs_url(project: Project):
+    if project.file_storages == missing:
+        raise REPError(f"The project object has no file storages information.")
+    rest_gateways = [fs for fs in project.file_storages if fs["obj_type"] == "RestGateway"]
+    rest_gateways.sort(key=lambda fs: fs["priority"], reverse=True)
+
+    if not rest_gateways:
+        raise REPError(f"Project {project.name} (id={project.id}) has no Rest Gateway defined.")
+
+    for d in rest_gateways:
+        url = d["url"]
+        try:
+            r = requests.get(url, verify=False, timeout=2)
+        except Exception as ex:
+            log.debug(ex)
+            continue
+        if r.status_code == 200:
+            return url
+    return None

ansys/rep/client/jms/resource/.gitignore

@@ -0,0 +1 @@
+*_base.py