Admin features new (#119)

* Adds admin features

* Updates submodule

* Adds archiving mechanism for admin query

* Adds limits to orga and changes mutation to change them

* Updates submodule

* Resolves PR comments

* Adds task types enums

* Removes type based task cancel mutations

* Adds limits

* Resolves PR comments

* Resolves PR comments

* Small fix

* Updates submodule

* Updates submodule

* Small change

* Resolves PR comments

* Updates submodule

* Updates submodule

* Merge submodules

---------

Co-authored-by: Lina <[email protected]>

Author: SimonDegrafKern

alembic/versions/3b118e1e02cb_adds_limits_to_orga.py

@@ -0,0 +1,32 @@
+"""Adds limits to orga
+
+Revision ID: 3b118e1e02cb
+Revises: a5c341d1a7ef
+Create Date: 2023-03-01 10:49:53.348290
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '3b118e1e02cb'
+down_revision = 'a5c341d1a7ef'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('organization', sa.Column('max_rows', sa.Integer(), nullable=True))
+    op.add_column('organization', sa.Column('max_cols', sa.Integer(), nullable=True))
+    op.add_column('organization', sa.Column('max_char_count', sa.Integer(), nullable=True))
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('organization', 'max_char_count')
+    op.drop_column('organization', 'max_cols')
+    op.drop_column('organization', 'max_rows')
+    # ### end Alembic commands ###

alembic/versions/a5c341d1a7ef_adds_admin_features.py

@@ -0,0 +1,47 @@
+"""Adds admin features
+
+Revision ID: a5c341d1a7ef
+Revises: 0e0e4aeac7eb
+Create Date: 2023-02-22 08:43:54.162780
+
+"""
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision = 'a5c341d1a7ef'
+down_revision = '0e0e4aeac7eb'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('admin_message',
+    sa.Column('id', postgresql.UUID(as_uuid=True), nullable=False),
+    sa.Column('text', sa.String(), nullable=True),
+    sa.Column('level', sa.String(), nullable=True),
+    sa.Column('archived', sa.Boolean(), nullable=True),
+    sa.Column('created_at', sa.DateTime(), nullable=True),
+    sa.Column('created_by', postgresql.UUID(as_uuid=True), nullable=True),
+    sa.Column('archive_date', sa.DateTime(), nullable=True),
+    sa.Column('archived_by', postgresql.UUID(as_uuid=True), nullable=True),
+    sa.Column('archived_reason', sa.String(), nullable=True),
+    sa.ForeignKeyConstraint(['archived_by'], ['user.id'], ),
+    sa.ForeignKeyConstraint(['created_by'], ['user.id'], ),
+    sa.PrimaryKeyConstraint('id')
+    )
+    op.create_index(op.f('ix_admin_message_archived_by'), 'admin_message', ['archived_by'], unique=False)
+    op.create_index(op.f('ix_admin_message_created_by'), 'admin_message', ['created_by'], unique=False)
+    op.add_column('user', sa.Column('last_interaction', sa.DateTime(), nullable=True))
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('user', 'last_interaction')
+    op.drop_index(op.f('ix_admin_message_created_by'), table_name='admin_message')
+    op.drop_index(op.f('ix_admin_message_archived_by'), table_name='admin_message')
+    op.drop_table('admin_message')
+    # ### end Alembic commands ###

controller/admin_message/manager.py

@@ -0,0 +1,53 @@
+from typing import List
+from datetime import datetime
+from submodules.model.business_objects import admin_message, general
+from submodules.model.models import AdminMessage
+from util.notification import send_global_update_for_all_organizations
+
+
+def get_messages(limit: int = 100, active_only: bool = True) -> List[AdminMessage]:
+    messages = admin_message.get_all_active(limit)
+    message_archived = False
+    messages_to_return = []
+    now = datetime.now()
+    for message in messages:
+        if now > message.archive_date:
+            admin_message.archive(
+                message.id, None, now, "Archive date reached.", with_commit=False
+            )
+            message_archived = True
+        elif active_only:
+            messages_to_return.append(message)
+
+    if message_archived:
+        general.commit()
+        send_global_update_for_all_organizations("admin_message")
+
+    if not active_only:
+        messages_to_return = admin_message.get_all(limit)
+    return messages_to_return
+
+
+def create_admin_message(
+    text: str, level: str, archive_date: datetime, created_by: str
+) -> AdminMessage:
+    now = datetime.now().astimezone(archive_date.tzinfo)
+
+    if archive_date < now:
+        raise ValueError("Archive date not valid")
+    return admin_message.create(
+        text=text,
+        level=level,
+        archive_date=archive_date,
+        created_by=created_by,
+        with_commit=True,
+    )
+
+
+def archive_admin_message(
+    message_id: str, archived_by: str, archived_reason: str
+) -> None:
+    archive_date = datetime.now()
+    admin_message.archive(
+        message_id, archived_by, archive_date, archived_reason, with_commit=True
+    )

controller/monitor/manager.py

@@ -0,0 +1,49 @@
+from typing import Any, List
+from submodules.model.business_objects import monitor as task_monitor
+
+
+def monitor_all_tasks(project_id: str = None, only_running: bool = True) -> List[Any]:
+    return task_monitor.get_all_tasks(project_id, only_running)
+
+
+def cancel_all_running_tasks(project_id: str = None) -> None:
+    task_monitor.cancel_all_running_tasks(project_id)
+
+
+def cancel_upload_task(project_id: str = None, upload_task_id: str = None) -> None:
+    task_monitor.set_upload_task_to_failed(project_id, upload_task_id, with_commit=True)
+
+
+def cancel_weak_supervision(project_id: str = None, payload_id: str = None) -> None:
+    task_monitor.set_weak_supervision_to_failed(
+        project_id, payload_id, with_commit=True
+    )
+
+
+def cancel_attribute_calculation(
+    project_id: str = None, attribute_id: str = None
+) -> None:
+    task_monitor.set_attribute_calculation_to_failed(
+        project_id, attribute_id, with_commit=True
+    )
+
+
+def cancel_embedding(project_id: str = None, embedding_id: str = None) -> None:
+    task_monitor.set_embedding_to_failed(project_id, embedding_id, with_commit=True)
+
+
+def cancel_information_source_payload(
+    project_id: str = None, payload_id: str = None
+) -> None:
+    task_monitor.set_information_source_payloads_to_failed(
+        project_id, payload_id, with_commit=True
+    )
+
+
+def cancel_record_tokenization_task(
+    project_id: str = None,
+    tokenization_task_id: str = None,
+) -> None:
+    task_monitor.set_record_tokenization_task_to_failed(
+        project_id, tokenization_task_id, with_commit=True
+    )

controller/user/manager.py

@@ -1,20 +1,25 @@
 from typing import Dict
 from submodules.model import User, enums
-from submodules.model.business_objects import user
-from submodules.model.business_objects import general
+from submodules.model.business_objects import user, user_activity, general
 from controller.auth import kratos
 from submodules.model.exceptions import EntityNotFoundException
 from controller.organization import manager as organization_manager
+from datetime import datetime, timedelta
+from util.decorator import param_throttle
 
 
 def get_user(user_id: str) -> User:
-    return user.get(user_id)
+    user_item = user.get(user_id)
+    if user_item:
+        update_last_interaction(user_item.id)
+    return user_item
 
 
 def get_or_create_user(user_id: str) -> User:
     user_item = user.get(user_id)
     if not user_item:
         user_item = user.create(user_id, with_commit=True)
+    update_last_interaction(user_item.id)
     return user_item
 
 
@@ -74,3 +79,16 @@ def remove_organization_from_user(user_mail: str) -> None:
         raise Exception("User has no organization")
 
     user.remove_organization(user_id, with_commit=True)
+
+
+def get_active_users(minutes: str, order_by_interaction: bool) -> User:
+    now = datetime.now()
+    last_interaction_range = (now - timedelta(minutes=minutes)) if minutes else None
+    return user_activity.get_active_users_in_range(
+        last_interaction_range, order_by_interaction
+    )
+
+
+@param_throttle(seconds=10)
+def update_last_interaction(user_id: str) -> None:
+    user_activity.update_last_interaction(user_id)

graphql_api/mutation/admin_message.py

@@ -0,0 +1,43 @@
+from datetime import datetime
+from controller.auth import manager as auth
+from controller.admin_message import manager
+import graphene
+from util.notification import send_global_update_for_all_organizations
+
+
+class CreateAdminMessage(graphene.Mutation):
+    class Arguments:
+        text = graphene.String(required=True)
+        level = graphene.String(required=True)
+        archive_date = graphene.DateTime(required=True)
+
+    ok = graphene.Boolean()
+
+    def mutate(self, info, text: str, level: str, archive_date: datetime):
+        auth.check_demo_access(info)
+        auth.check_admin_access(info)
+        user_id = auth.get_user_id_by_info(info)
+        manager.create_admin_message(text, level, archive_date, user_id)
+        send_global_update_for_all_organizations("admin_message")
+        return CreateAdminMessage(ok=True)
+
+
+class ArchiveAdminMessage(graphene.Mutation):
+    class Arguments:
+        message_id = graphene.ID(required=True)
+        archived_reason = graphene.String(required=True)
+
+    ok = graphene.Boolean()
+
+    def mutate(self, info, message_id: str, archived_reason: str):
+        auth.check_demo_access(info)
+        auth.check_admin_access(info)
+        user_id = auth.get_user_id_by_info(info)
+        manager.archive_admin_message(message_id, user_id, archived_reason)
+        send_global_update_for_all_organizations(f"admin_message")
+        return ArchiveAdminMessage(ok=True)
+
+
+class AdminMessageMutation(graphene.ObjectType):
+    create_admin_message = CreateAdminMessage.Field()
+    archive_admin_message = ArchiveAdminMessage.Field()

graphql_api/mutation/monitor.py

@@ -0,0 +1,57 @@
+import graphene
+from controller.auth import manager as auth
+from controller.monitor import manager
+from submodules.model import enums
+
+
+class CancelTask(graphene.Mutation):
+    class Arguments:
+        project_id = graphene.ID(required=True)
+        task_id = graphene.ID(required=True)
+        task_type = graphene.ID(required=True)
+
+    ok = graphene.Boolean()
+
+    def mutate(self, info, project_id: str, task_id: str, task_type: str):
+        auth.check_admin_access(info)
+        if task_type == enums.TaskType.ATTRIBUTE_CALCULATION.value:
+            manager.cancel_attribute_calculation(project_id, task_id)
+        elif task_type == enums.TaskType.EMBEDDING.value:
+            manager.cancel_embedding(project_id, task_id)
+        elif task_type == enums.TaskType.INFORMATION_SOURCE.value:
+            manager.cancel_information_source_payload(project_id, task_id)
+        elif task_type == enums.TaskType.TOKENIZATION.value:
+            manager.cancel_record_tokenization_task(project_id, task_id)
+        elif task_type == enums.TaskType.UPLOAD_TASK.value:
+            manager.cancel_upload_task(project_id, task_id)
+        elif task_type == enums.TaskType.WEAK_SUPERVISION.value:
+            manager.cancel_weak_supervision(project_id, task_id)
+        else:
+            raise ValueError(f"{task_type} is no valid task type")
+        return CancelTask(ok=True)
+
+
+class CancelAllRunningTasks(graphene.Mutation):
+    ok = graphene.Boolean()
+
+    def mutate(self, info):
+        auth.check_admin_access(info)
+        manager.cancel_all_running_tasks()
+        return CancelAllRunningTasks(ok=True)
+
+
+class CancelInformationSourcePayload(graphene.Mutation):
+    class Arguments:
+        payload_id = graphene.ID(required=True)
+
+    ok = graphene.Boolean()
+
+    def mutate(self, info, payload_id: str):
+        auth.check_admin_access(info)
+        manager.cancel_information_source_payload(payload_id=payload_id)
+        return CancelInformationSourcePayload(ok=True)
+
+
+class MonitorMutation(graphene.ObjectType):
+    cancel_task = CancelTask.Field()
+    cancel_all_running_tasks = CancelAllRunningTasks.Field()

graphql_api/mutation/organization.py

@@ -99,7 +99,8 @@ class Arguments:
 
     def mutate(self, info, org_id: str, changes: Dict[str, Any]):
         auth.check_demo_access(info)
-        auth.check_admin_access(info)
+        if config_service.get_config_value("is_managed"):
+            auth.check_admin_access(info)
         organization_manager.change_organization(org_id, changes)
         return DeleteOrganization(ok=True)
 

graphql_api/query/admin_message.py

@@ -0,0 +1,24 @@
+import graphene
+
+from controller.auth import manager as auth
+from graphql_api.types import AdminMessage
+from controller.admin_message import manager
+
+
+class AdminMessageQuery(graphene.ObjectType):
+    all_admin_messages = graphene.Field(
+        graphene.List(AdminMessage), limit=graphene.Int()
+    )
+
+    all_active_admin_messages = graphene.Field(
+        graphene.List(AdminMessage), limit=graphene.Int()
+    )
+
+    def resolve_all_admin_messages(self, info, limit: int = 100) -> AdminMessage:
+        auth.check_demo_access(info)
+        auth.check_admin_access(info)
+        return manager.get_messages(limit, active_only=False)
+
+    def resolve_all_active_admin_messages(self, info, limit: int = 100) -> AdminMessage:
+        auth.check_demo_access(info)
+        return manager.get_messages(limit, active_only=True)

graphql_api/query/monitor.py

@@ -0,0 +1,12 @@
+import graphene
+from graphql_api.types import Task
+from controller.auth import manager as auth
+from controller.monitor import manager
+
+
+class MonitorQuery(graphene.ObjectType):
+    all_tasks = graphene.Field(graphene.List(Task), only_running=graphene.Boolean())
+
+    def resolve_all_tasks(self, info, only_running: bool = True):
+        auth.check_admin_access(info)
+        return manager.monitor_all_tasks(only_running=only_running)