+
diff --git a/src/app/types/bulks.ts b/src/app/types/bulks.ts
index 39c4a92a..2baf26a7 100644
--- a/src/app/types/bulks.ts
+++ b/src/app/types/bulks.ts
@@ -1,6 +1,6 @@
type StatusType = 'create' | 'update' | 'delete' | 'skip' | 'error'
-interface ValidationResult {
+interface EachResult {
row: number
id: string
status: StatusType
@@ -18,8 +18,8 @@ interface MissingUser {
groups: string[]
}
-interface ValidationSummary {
- items: ValidationResult[]
+interface ValidationResults {
+ results: EachResult[]
summary: Summary
missingUsers: MissingUser[]
total: number
@@ -27,17 +27,6 @@ interface ValidationSummary {
pageSize: number
}
-interface UploadResult {
- row: number
- id: string
- status: StatusType
- userName: string
- eppn: string[]
- emails: string[]
- groups: string[]
- code?: string
-}
-
interface Summary {
create: number
update: number
@@ -46,8 +35,8 @@ interface Summary {
error: number
}
-interface ResultSummary {
- results: UploadResult[]
+interface ExecuteResults {
+ results: EachResult[]
summary: Summary
fileInfo: {
fileName: string
@@ -92,6 +81,7 @@ interface BulkIndicator {
icon: string
key: StatusType
}
-export type { StatusType, ValidationResult, MissingUser, UploadResult, Summary, ExcuteResponse,
- BulkProcessingStatus, ValidationSummary, ResultSummary, ExcuteRequest, UploadQuery, BulkIndicator,
+export type { StatusType, EachResult, MissingUser, Summary, ExcuteResponse,
+ BulkProcessingStatus, ValidationResults, ExecuteResults, ExcuteRequest,
+ UploadQuery, BulkIndicator,
}
diff --git a/src/app/types/history.ts b/src/app/types/history.ts
index 352788cc..e8a7471a 100644
--- a/src/app/types/history.ts
+++ b/src/app/types/history.ts
@@ -9,6 +9,7 @@ interface DownloadHistoryData {
parentId: string | undefined
filePath: string
fileId: string
+ fileExists: boolean
repositoryCount: number
groupCount: number
userCount: number
@@ -56,13 +57,6 @@ interface HistoryQuery {
i?: string[]
}
-interface DownloadGroupItem {
- parent: DownloadHistoryData
- children: DownloadHistoryData[]
- hasMoreChildren: boolean
- childrenLimit: number
-}
-
interface TableConfig {
enableExpand?: boolean
showStatus?: boolean
@@ -89,8 +83,8 @@ interface PublicStatusUpdateRequest {
public: boolean
}
-type ActionRow = DownloadGroupItem | UploadHistoryData
+type ActionRow = DownloadHistoryData | UploadHistoryData
export type { DownloadHistoryData, UploadHistoryData, DownloadApiModel, UploadApiModel,
HistoryQuery, TableConfig, FilterOptionsResponse, SelectOption, StatusConfig,
- PublicStatusUpdateRequest, ActionRow, DownloadGroupItem }
+ PublicStatusUpdateRequest, ActionRow }
diff --git a/src/server/api/bulk.py b/src/server/api/bulk.py
index 2a1fa9b8..086d6910 100644
--- a/src/server/api/bulk.py
+++ b/src/server/api/bulk.py
@@ -4,18 +4,28 @@
"""API router for bulk endpoints."""
-import typing as t
+from uuid import UUID
-from pathlib import Path
-from uuid import UUID, uuid7
-
-from flask import Blueprint
-from flask_login import current_user
+from flask import Blueprint, current_app
+from flask_login import current_user, login_required
from flask_pydantic import validate
-from redis.exceptions import ConnectionError as RedisConnectionError
-from server.api.helpers import validate_files
-from server.api.schemas import (
+from server.auth import is_user_logged_in
+from server.const import DEFAULT_SEARCH_COUNT, USER_ROLES
+from server.entities.bulk import ExecuteResults, ValidateResults
+from server.exc import (
+ FileFormatError,
+ FileNotFound,
+ FileValidationError,
+ RecordNotFound,
+ TaskExcutionError,
+)
+from server.messages import E
+from server.services import bulks, history_table, repositories
+from server.services.utils import get_permitted_repository_ids
+
+from .helpers import roles_required, validate_files
+from .schemas import (
BulkBody,
BulkFileForm,
ErrorResponse,
@@ -23,17 +33,17 @@
TargetRepositoryForm,
UploadQuery,
)
-from server.config import config
-from server.entities.bulk import ResultSummary, ValidateSummary
-from server.entities.login_user import LoginUser
-from server.exc import RecordNotFound
-from server.services import bulks, history_table
+
+
+STATUS_MAP = {0: "create", 1: "update", 2: "delete", 3: "skip", 4: "error"}
bp = Blueprint("bulk", __name__)
@bp.post("/upload-file")
+@login_required
+@roles_required(USER_ROLES.SYSTEM_ADMIN, USER_ROLES.REPOSITORY_ADMIN)
@validate_files
@validate(response_by_alias=True)
def upload_file(
@@ -49,28 +59,28 @@ def upload_file(
BulkBody: The response containing task ID
ErrorResponse: The response containing task ID or error message.
"""
- temp_id = uuid7()
- temp_dir = Path(config.STORAGE.local.temporary)
- original_filename = files.bulk_file.filename or "upload_file"
- operator_id = t.cast("LoginUser", current_user).map_id
- operator_name = t.cast("LoginUser", current_user).user_name
- new_filename = f"{temp_id}_{Path(original_filename).name}"
- file_path = temp_dir / new_filename
- files.bulk_file.save(str(file_path))
- file_content = {"repositories": [{"id": form.repository_id}]}
- history_table.create_file(
- file_id=temp_id, file_path=str(file_path), file_content=file_content
- )
-
- bulks.delete_temporary_file.apply_async((str(temp_id),), countdown=3600)
+ if repositories.get_by_id(form.repository_id) is None:
+ error = E.REPOSITORY_NOT_FOUND % {"id": form.repository_id}
+ current_app.logger.error(error)
+ return ErrorResponse(message=error), 404
+ if (
+ not current_user.is_system_admin
+ and form.repository_id not in get_permitted_repository_ids()
+ ):
+ error = E.REPOSITORY_FORBIDDEN % {"id": form.repository_id}
+ current_app.logger.error(error)
+ return ErrorResponse(message=error), 403
+ temp_file_id = bulks.upload_file(form.repository_id, files.bulk_file)
task = bulks.validate_upload_data.apply_async(
- (operator_id, operator_name, temp_id),
+ (current_user.map_id, current_user.user_name, temp_file_id),
session_required=True, # pyright: ignore[reportCallIssue]
)
- return BulkBody(task_id=task.id, temp_file_id=temp_id), 202
+ return BulkBody(task_id=task.id, temp_file_id=temp_file_id), 200
@bp.get("/validate/status/
")
+@login_required
+@roles_required(USER_ROLES.SYSTEM_ADMIN, USER_ROLES.REPOSITORY_ADMIN)
@validate(response_by_alias=True)
def validate_status(task_id: str) -> tuple[BulkBody | ErrorResponse, int]:
"""Get the status of a validation task.
@@ -82,23 +92,22 @@ def validate_status(task_id: str) -> tuple[BulkBody | ErrorResponse, int]:
BulkBody: The response containing task status
ErrorResponse: The response containing task status or error message
"""
- try:
- res = bulks.validate_upload_data.AsyncResult(task_id)
- except RedisConnectionError:
- error = f"Failed to connect to Redis: {task_id}"
- return ErrorResponse(code="", message=error), 500
+ res = bulks.get_validate_task_result(task_id)
if not res:
- error = f"Task not found: {task_id}"
- return ErrorResponse(code="", message=error), 404
+ error = E.TASK_NOT_FOUND % {"task_id": task_id}
+ current_app.logger.error(error)
+ return ErrorResponse(message=error), 404
return BulkBody(status=res.state), 200
@bp.get("/validate/result/")
+@login_required
+@roles_required(USER_ROLES.SYSTEM_ADMIN, USER_ROLES.REPOSITORY_ADMIN)
@validate(response_by_alias=True)
def validate_result(
query: UploadQuery,
task_id: str,
-) -> tuple[ValidateSummary | ErrorResponse, int]:
+) -> tuple[ValidateResults | ErrorResponse, int]:
"""Get the result of a validation task.
Args:
@@ -110,36 +119,37 @@ def validate_result(
ErrorResponse: The response containing validation result or error message.
"""
try:
- res = bulks.validate_upload_data.AsyncResult(task_id)
- except RedisConnectionError:
- error = f"Failed to connect to Redis: {task_id}"
- return ErrorResponse(code="", message=error), 500
- if not res:
- error = f"Task not found: {task_id}"
- return ErrorResponse(code="", message=error), 404
- if not res.successful():
- error = f"Task not successful: {task_id}"
- return ErrorResponse(code="", message=error), 400
- history_id = res.result
- if isinstance(history_id, BaseException):
- error = f"Task resulted in an exception: {history_id}"
- return ErrorResponse(code="", message=error), 400
- status_filter = (
- [
- {0: "create", 1: "update", 2: "delete", 3: "skip", 4: "error"}[status]
- for status in query.f
- ]
- if query.f
- else []
- )
- offset = query.p or 1
- size = query.l or 20
- return bulks.get_validate_result(
- history_id=history_id, status_filter=status_filter, offset=offset, size=size
- ), 200
+ res = bulks.get_validate_task_result(task_id)
+ match res.result:
+ case FileNotFound():
+ return ErrorResponse(message=res.result.message), 404
+ case FileValidationError() | FileFormatError():
+ return ErrorResponse(message=res.result.message), 400
+ case UUID():
+ pass
+ case _:
+ return ErrorResponse(message=E.UNEXPECTED_SERVER_ERROR), 500
+ history_id = res.result
+ status_filter = [STATUS_MAP[status] for status in query.f] if query.f else []
+ offset = query.p or 1
+ size = query.l or DEFAULT_SEARCH_COUNT
+ if not is_user_logged_in(
+ current_user
+ ) or not bulks.chack_permission_to_operation(history_id, current_user.map_id):
+ error = E.OPERATION_FORBIDDEN
+ current_app.logger.error(error)
+ return ErrorResponse(message=error), 403
+ result = bulks.get_validate_result(
+ history_id=history_id, status_filter=status_filter, offset=offset, size=size
+ )
+ except (RecordNotFound, TaskExcutionError) as exc:
+ return ErrorResponse(message=exc.message), 404
+ return result, 200
@bp.post("/execute")
+@login_required
+@roles_required(USER_ROLES.SYSTEM_ADMIN, USER_ROLES.REPOSITORY_ADMIN)
@validate(response_by_alias=True)
def execute(body: ExcuteRequest) -> tuple[BulkBody | ErrorResponse, int]:
"""Execute a bulk upload.
@@ -155,6 +165,12 @@ def execute(body: ExcuteRequest) -> tuple[BulkBody | ErrorResponse, int]:
"""
try:
history_id = history_table.get_history_by_file_id(body.temp_file_id).id
+ if not is_user_logged_in(
+ current_user
+ ) or not bulks.chack_permission_to_operation(history_id, current_user.map_id):
+ error = E.OPERATION_FORBIDDEN
+ current_app.logger.error(error)
+ return ErrorResponse(message=error), 403
task = bulks.update_users.apply_async(
kwargs={
"history_id": history_id,
@@ -163,12 +179,14 @@ def execute(body: ExcuteRequest) -> tuple[BulkBody | ErrorResponse, int]:
},
)
except RecordNotFound as exc:
- return ErrorResponse(code="", message=str(exc)), 404
+ return ErrorResponse(message=exc.message), 404
return BulkBody(task_id=task.id, history_id=history_id), 200
@bp.get("/execute/status/")
-@validate()
+@login_required
+@roles_required(USER_ROLES.SYSTEM_ADMIN, USER_ROLES.REPOSITORY_ADMIN)
+@validate(response_by_alias=True)
def execute_status(task_id: str) -> tuple[BulkBody | ErrorResponse, int]:
"""Get the status of an execution task.
@@ -180,21 +198,19 @@ def execute_status(task_id: str) -> tuple[BulkBody | ErrorResponse, int]:
ErrorResponse: The response containing task status or error message
"""
try:
- res = bulks.update_users.AsyncResult(task_id)
- except RedisConnectionError:
- error = f"Failed to connect to Redis: {task_id}"
- return ErrorResponse(code="", message=error), 500
- if not res:
- error = f"Task not found: {task_id}"
- return ErrorResponse(code="", message=error), 404
+ res = bulks.get_execute_task_result(task_id)
+ except TaskExcutionError as exc:
+ return ErrorResponse(message=exc.message), 404
return BulkBody(status=res.state), 200
@bp.get("/result/")
+@login_required
+@roles_required(USER_ROLES.SYSTEM_ADMIN, USER_ROLES.REPOSITORY_ADMIN)
@validate(response_by_alias=True)
def result(
history_id: UUID, query: UploadQuery
-) -> tuple[ResultSummary | ErrorResponse, int]:
+) -> tuple[ExecuteResults | ErrorResponse, int]:
"""Get the result of a bulk upload.
Args:
@@ -202,24 +218,21 @@ def result(
query(UploadQuery): Query parameters for filtering results.
Returns:
- ResultSummary: Summary of displayed history If the get is successful
+ ExecuteResults: Summary of displayed history If the get is successful
ErrorResponse: If the get is failed
"""
- status_filter = (
- [
- {0: "create", 1: "update", 2: "delete", 3: "skip", 4: "error"}[status]
- for status in query.f
- ]
- if query.f
- else []
- )
+ status_filter = [STATUS_MAP[status] for status in query.f] if query.f else []
offset = query.p or 1
- size = query.l or 10
+ size = query.l or DEFAULT_SEARCH_COUNT
try:
+ if not bulks.chack_permission_to_view(history_id):
+ error = E.OPERATION_FORBIDDEN
+ current_app.logger.error(error)
+ return ErrorResponse(message=error), 403
result = bulks.get_upload_result(
history_id=history_id, status_filter=status_filter, offset=offset, size=size
)
except RecordNotFound as exc:
- return ErrorResponse(code="", message=str(exc)), 404
+ return ErrorResponse(message=exc.message), 404
return result, 200
diff --git a/src/server/api/history.py b/src/server/api/history.py
index 2a5c2824..32e0d391 100644
--- a/src/server/api/history.py
+++ b/src/server/api/history.py
@@ -16,7 +16,8 @@
from server.const import USER_ROLES
from server.entities.history_detail import HistoryQuery
from server.entities.search_request import FilterOption, SearchResult
-from server.exc import DatabaseError, InvalidQueryError, RecordNotFound
+from server.exc import InvalidQueryError, RecordNotFound
+from server.messages import E
from server.services import history
from server.services.utils import search_history_filter_options
@@ -59,8 +60,8 @@ def filter_options_operators(
"""
try:
result = history.get_filter_items(tab, key="o", criteria=query)
- except InvalidQueryError as ex:
- return ErrorResponse(code="", message=str(ex)), 400
+ except InvalidQueryError as exc:
+ return ErrorResponse(message=exc.message), 400
return result, 200
@@ -81,15 +82,10 @@ def get(
SearchResult: if successful and status code 200
ErrorResponse: if a connection error occurs and status code 503
"""
- try:
- if tab == "download":
- result = history.get_download_history_data(query)
- else:
- result = history.get_upload_history_data(query)
- except DatabaseError:
- error = f"{tab} table connection error"
- current_app.logger.error(error)
- return ErrorResponse(code="", message=error), 503
+ if tab == "download":
+ result = history.get_download_history_data(query)
+ else:
+ result = history.get_upload_history_data(query)
return result, 200
@@ -115,8 +111,8 @@ def public_status(
result: bool = history.update_public_status(
tab=tab, history_id=history_id, public=body.public
)
- except RecordNotFound as ex:
- return ErrorResponse(code="", message=str(ex)), 404
+ except RecordNotFound as exc:
+ return ErrorResponse(message=exc.message), 404
return HistoryPublic(public=result), 200
@@ -136,12 +132,12 @@ def files(file_id: UUID) -> Response | tuple[ErrorResponse, int]:
try:
path_str = history.get_file_path(file_id)
file_path = Path(path_str)
- except RecordNotFound as ex:
- return ErrorResponse(code="", message=str(ex)), 404
+ except RecordNotFound as exc:
+ return ErrorResponse(message=exc.message), 404
if not file_path.exists():
- error = f"File not found: {file_id}"
+ error = E.FILE_NOT_FOUND % {"path": file_path}
current_app.logger.error(error)
- return ErrorResponse(code="", message=error), 404
+ return ErrorResponse(message=error), 404
return send_file(path_or_file=file_path)
@@ -160,8 +156,8 @@ def is_exist_files(file_id: UUID) -> tuple[bool | ErrorResponse, int]:
"""
try:
file_path = Path(history.get_file_path(file_id))
- except RecordNotFound as ex:
- return ErrorResponse(code="", message=str(ex)), 404
+ except RecordNotFound as exc:
+ return ErrorResponse(message=exc.message), 404
if not Path(file_path).exists():
return False, 200
return True, 200
diff --git a/src/server/api/schemas.py b/src/server/api/schemas.py
index c3e63630..193dfe28 100644
--- a/src/server/api/schemas.py
+++ b/src/server/api/schemas.py
@@ -354,3 +354,23 @@ class UploadQuery(BaseModel):
l: t.Annotated[int | None, "length"] = None # noqa: E741
"""Page size (number of items per page)."""
+
+
+class FileQuery(BaseModel):
+ """Query parameters for file export."""
+
+ f: t.Annotated[t.Literal["tsv", "csv"], "format"] = "tsv"
+ """File format for export."""
+
+ model_config = ignore_extra_config
+ """Configure to ignore extra fields."""
+
+
+class ExportBody(BaseModel):
+ """Body for user export request."""
+
+ user_ids: list[str]
+ """List of user IDs to export."""
+
+ model_config = camel_case_config
+ """Configure to use camelCase aliasing."""
diff --git a/src/server/api/users.py b/src/server/api/users.py
index 17345031..9e17b026 100644
--- a/src/server/api/users.py
+++ b/src/server/api/users.py
@@ -9,7 +9,7 @@
import traceback
import typing as t
-from flask import Blueprint, current_app, url_for
+from flask import Blueprint, Response, current_app, send_file, url_for
from flask_login import current_user, login_required
from flask_pydantic import validate
@@ -18,6 +18,7 @@
from server.entities.search_request import FilterOption, SearchResult
from server.entities.user_detail import UserDetail
from server.exc import (
+ InvalidExportError,
InvalidFormError,
InvalidQueryError,
RequestConflict,
@@ -33,7 +34,7 @@
from .auth import logout
from .helpers import roles_required
-from .schemas import ErrorResponse, UsersQuery
+from .schemas import ErrorResponse, ExportBody, FileQuery, UsersQuery
bp = Blueprint("users", __name__)
@@ -203,3 +204,31 @@ def filter_options() -> list[FilterOption]:
list[FilterOption]: List of filter options for user search.
"""
return search_users_options()
+
+
+@bp.get("/export")
+@bp.post("/export")
+@login_required
+@roles_required(USER_ROLES.SYSTEM_ADMIN, USER_ROLES.REPOSITORY_ADMIN)
+@validate(response_by_alias=True)
+def user_export(
+ body: ExportBody, query: FileQuery
+) -> Response | tuple[ErrorResponse, int]:
+ """Export users to a file for bulk processing.
+
+ Args:
+ body (ExportBody):
+ The body of the export request containing the IDs of the users to export.
+ query (FileQuery): The query parameters for the export.
+
+ Returns:
+ Response: The response containing the exported file
+ ErrorResponse: The response containing an error message if the export fails
+ """
+ try:
+ files = users.make_export_file(
+ body.user_ids or [], query, current_user.map_id, current_user.name
+ )
+ except InvalidExportError as exc:
+ return ErrorResponse(message=exc.message), 403
+ return send_file(files)
diff --git a/src/server/cli/app.py b/src/server/cli/app.py
new file mode 100644
index 00000000..2af4be07
--- /dev/null
+++ b/src/server/cli/app.py
@@ -0,0 +1,26 @@
+#
+# Copyright (C) 2025 National Institute of Informatics.
+#
+
+"""Application command-line interface."""
+
+import pathlib
+import tomllib
+
+import click
+
+from flask import current_app
+
+
+@click.group()
+def app() -> None:
+ """Manage application."""
+
+
+@app.command()
+def version() -> None:
+ """Display application version."""
+ with pathlib.Path("pyproject.toml").open("rb") as f:
+ pyproject = tomllib.load(f)
+ version = pyproject["project"]["version"]
+ current_app.logger.info(version)
diff --git a/src/server/clients/groups.py b/src/server/clients/groups.py
index 3b2eac2b..6fca8573 100644
--- a/src/server/clients/groups.py
+++ b/src/server/clients/groups.py
@@ -91,10 +91,6 @@ def search(
by_alias=True,
)
- from contrib import dump
-
- dump(auth_params | attributes_params | query_params, "groups_search_query")
-
response = requests.get(
f"{config.MAP_CORE.base_url}{MAP_GROUPS_ENDPOINT}",
params=auth_params | attributes_params | query_params,
@@ -104,8 +100,6 @@ def search(
timeout=config.MAP_CORE.timeout,
)
- dump(response.text, "groups_search_response")
-
if response.status_code > HTTPStatus.BAD_REQUEST:
response.raise_for_status()
@@ -216,10 +210,6 @@ def post(
alias_generator(name) for name in exclude
])
- from contrib import dump
-
- dump(auth_params | payload, "groups_post_payload")
-
response = requests.post(
f"{config.MAP_CORE.base_url}{MAP_GROUPS_ENDPOINT}",
params=attributes_params,
@@ -230,8 +220,6 @@ def post(
timeout=config.MAP_CORE.timeout,
)
- dump(response.text, "groups_post_response")
-
if response.status_code > HTTPStatus.BAD_REQUEST:
response.raise_for_status()
diff --git a/src/server/clients/services.py b/src/server/clients/services.py
index 0dc18ceb..a6c0e4f3 100644
--- a/src/server/clients/services.py
+++ b/src/server/clients/services.py
@@ -214,10 +214,6 @@ def post(
alias_generator(name) for name in exclude
])
- from contrib import dump
-
- dump({"request": auth_params} | payload, "service_post_payload")
-
response = requests.post(
f"{config.MAP_CORE.base_url}{MAP_SERVICES_ENDPOINT}",
params=attributes_params,
@@ -228,8 +224,6 @@ def post(
timeout=config.MAP_CORE.timeout,
)
- dump(response.text, "service_post_response")
-
status_code = response.status_code
if status_code not in {HTTPStatus.BAD_REQUEST, HTTPStatus.CONFLICT}:
response.raise_for_status()
diff --git a/src/server/clients/users.py b/src/server/clients/users.py
index 4a4e419f..6f734811 100644
--- a/src/server/clients/users.py
+++ b/src/server/clients/users.py
@@ -91,10 +91,6 @@ def search(
by_alias=True,
)
- from contrib import dump
-
- dump(auth_params | attributes_params | query_params, "users_search_query")
-
response = requests.get(
f"{config.MAP_CORE.base_url}{MAP_USERS_ENDPOINT}",
params=auth_params | attributes_params | query_params,
@@ -104,8 +100,6 @@ def search(
timeout=config.MAP_CORE.timeout,
)
- dump(response.text, "users_search_response")
-
if response.status_code > HTTPStatus.BAD_REQUEST:
response.raise_for_status()
diff --git a/src/server/config.py b/src/server/config.py
index efcd5351..f9f6b2d6 100644
--- a/src/server/config.py
+++ b/src/server/config.py
@@ -83,6 +83,9 @@ class RuntimeConfig(BaseSettings):
GROUPS: GroupsConfig
"""Group related configuration values."""
+ USERS: UsersConfig
+ """Users related configuration values."""
+
POSTGRES: PostgresConfig = Field(
default_factory=lambda: PostgresConfig(), # noqa: PLW0108
exclude=True,
@@ -408,6 +411,22 @@ def __getitem__(self, key: USER_ROLES) -> str: # noqa: D105
return getattr(self, key)
+class UsersConfig(BaseModel):
+ """Schema for user export file configuration."""
+
+ export_fields: list[str] = [
+ "id",
+ "user_name",
+ "groups[].id",
+ "groups[].name",
+ "role",
+ "edu_person_principal_names[]",
+ "preferred_language",
+ "emails[]",
+ ]
+ """List of fields to include in the exported user details."""
+
+
class MapCoreConfig(BaseModel):
"""Schema for mAP Core service configuration."""
diff --git a/src/server/entities/bulk.py b/src/server/entities/bulk.py
index d7b034e2..38fbcd78 100644
--- a/src/server/entities/bulk.py
+++ b/src/server/entities/bulk.py
@@ -35,13 +35,13 @@ class RepositoryMember(BaseModel):
"""The users belonging to the repository."""
-class ValidateSummary(BaseModel):
+class ValidateResults(BaseModel):
"""Model for summary of bulk validation result."""
- results: list[CheckResult]
+ results: list[EachResult]
"""The list of validation results for each user."""
- summary: HistorySummary
+ summary: ResultSummary
"""The summary of the validation operation."""
missing_user: list[UserDetail] = []
@@ -51,7 +51,7 @@ class ValidateSummary(BaseModel):
"""Configure camelCase aliasing and forbid extra fields."""
-class HistorySummary(BaseModel):
+class ResultSummary(BaseModel):
"""Summary of the history operation."""
create: int
@@ -69,7 +69,7 @@ class HistorySummary(BaseModel):
"""Configure camelCase aliasing and forbid extra fields."""
-class CheckResult(BaseModel):
+class EachResult(BaseModel):
"""Model for result of validation check for each user."""
id: str | None
@@ -97,13 +97,13 @@ class CheckResult(BaseModel):
"""Configure camelCase aliasing and forbid extra fields."""
-class ResultSummary(BaseModel):
+class ExecuteResults(BaseModel):
"""Model for summary of bulk upload result."""
- items: list[CheckResult]
+ items: list[EachResult]
"""The list of upload results for each user."""
- summary: HistorySummary
+ summary: ResultSummary
"""The summary of the upload operation."""
file_id: UUID
@@ -147,7 +147,7 @@ class Aggregated(t.TypedDict):
summary: dict[str, int]
"""Summary of the aggregation."""
- results: list[CheckResult]
+ results: list[EachResult]
"""List of check results."""
missing_user: list[UserDetail]
diff --git a/src/server/entities/history_detail.py b/src/server/entities/history_detail.py
index 9a4c5108..58312570 100644
--- a/src/server/entities/history_detail.py
+++ b/src/server/entities/history_detail.py
@@ -46,6 +46,9 @@ class DownloadHistoryData(BaseModel):
file_path: str
"""Path of the downloaded file."""
+ file_exists: bool = False
+ """Indicates if the downloaded file still exists."""
+
repository_count: int
"""Number of repositories involved in the download."""
diff --git a/src/server/exc.py b/src/server/exc.py
index 0be07137..ea1d7b8f 100644
--- a/src/server/exc.py
+++ b/src/server/exc.py
@@ -193,3 +193,31 @@ class FileValidationError(BulkOperationError):
Errors caused by validation failures during bulk operations.
"""
+
+
+class FileNotFound(BulkOperationError): # noqa: N818
+ """Exception for file not found errors in bulk operations.
+
+ Errors caused by missing files during bulk operations.
+ """
+
+
+class FileFormatError(BulkOperationError):
+ """Exception for file format errors in bulk operations.
+
+ Errors caused by invalid file formats during bulk operations.
+ """
+
+
+class FileUploadError(BulkOperationError):
+ """Exception for file upload errors in bulk operations.
+
+ Errors caused by issues during file upload in bulk operations.
+ """
+
+
+class InvalidExportError(BulkOperationError):
+ """Exception for invalid export errors.
+
+ Errors caused by issues during export operations.
+ """
diff --git a/src/server/messages/error.py b/src/server/messages/error.py
index 885c53ed..bb336d6e 100644
--- a/src/server/messages/error.py
+++ b/src/server/messages/error.py
@@ -554,6 +554,20 @@
"System Administrator.",
)
+USER_CANNOT_EXPORT_SYSTEM_ADMIN = LogMessage(
+ "E361",
+ "System Administrator user is not allowed to be exported.",
+)
+
+USER_FORBIDDEN_EXPORT = LogMessage(
+ "E362",
+ "Logged-in user does not have permission to export a User.",
+)
+
+FAILED_CREATE_DOWNLOAD_HISTORY_RECORD = LogMessage(
+ "E363",
+ "Failed to create download history for file (id: %(file_id)s) in database.",
+)
UNAUTHORIZED = LogMessage(
"E401",
@@ -576,6 +590,125 @@
"The server application is currently unavailable.",
)
+FAILED_BULK_OPERATION = LogMessage(
+ "E600",
+ "Failed to perform bulk operation.",
+)
+
+FAILED_SAVE_UPLOADED_FILE = LogMessage(
+ "E601",
+ "Failed to save uploaded file (path: %(file_path)s).",
+)
+
+OPERATION_FORBIDDEN = LogMessage(
+ "E602",
+ "Logged-in user does not have permission to perform this operation.",
+)
+
+FAILED_GET_UPLOAD_HISTORY_RECORD = LogMessage(
+ "E610",
+ "Failed to get upload history (id: %(history_id)s) from database.",
+)
+
+FAILED_GET_UPLOAD_HISTORY_RECORD_BY_FILE_ID = LogMessage(
+ "E611",
+ "Failed to get upload history by file ID (file_id: %(file_id)s) from database.",
+)
+
+FAILED_CREATE_UPLOAD_HISTORY_RECORD = LogMessage(
+ "E612",
+ "Failed to create upload history for file (id: %(file_id)s) in database.",
+)
+
+FAILED_UPDATE_HISTORY_RECORD_STATUS = LogMessage(
+ "E613",
+ "Failed to update upload history status (id: %(history_id)s) in database.",
+)
+
+INVALID_UPLOAD_HISTORY_RECORD_ATTRIBUTES = LogMessage(
+ "E614",
+ "Results must include 'summary' and 'results' keys",
+)
+
+UPDATE_HISTORY_RECORD_NOT_FOUND = LogMessage(
+ "E615",
+ "Record (id: %(id)s) not found in database.",
+)
+
+FAILED_GET_FILE_RECORD = LogMessage(
+ "E616",
+ "Failed to get file record(file_id: %(file_id)s) from database.",
+)
+
+FAILED_DELETE_FILE_RECORD = LogMessage(
+ "E617",
+ "Failed to delete file record(file_id: %(file_id)s) from database.",
+)
+
+INVALID_Query = LogMessage(
+ "E618",
+ "offset: %(offset)s and size: %(size)s must be non-negative integers.",
+)
+
+FAILED_CREATE_FILE_RECORD = LogMessage(
+ "E620",
+ "Failed to create file record for file (id: %(file_id)s) in database.",
+)
+
+INVALID_FILE_STRUCTURE = LogMessage(
+ "E621",
+ "Invalid file structure.",
+)
+
+FILE_EXPIRED = LogMessage(
+ "E622", " File not found (path: %(path)s). It may have been expired."
+)
+
+FILE_FORMAT_UNSUPPORTED = LogMessage(
+ "E623",
+ "Unsupported file format (suffix: %(suffix)s).",
+)
+
+FILE_VALIDATION_ERROR = LogMessage(
+ "E624",
+ "File validation failed for task: %(task_id)s. Please check the file.",
+)
+
+TASK_NOT_FOUND = LogMessage(
+ "E634",
+ "Task (id: %(task_id)s) not found. It may have been expired.",
+)
+
+FAILED_GET_HISTORY_RECORDS = LogMessage(
+ "E700",
+ "Failed to get history records from table: %(table)s.",
+)
+
+FAILED_GET_HISTORY_RECORD = LogMessage(
+ "E701",
+ "Failed to get history record (id: %(history_id)s) from table: %(table)s.",
+)
+
+FAILED_UPDATE_PUBLIC = LogMessage(
+ "E702",
+ "Failed to update public status of history record (id: %(history_id)s) in"
+ " database.",
+)
+
+FILE_NOT_FOUND = LogMessage(
+ "E704",
+ "File not found (path: %(path)s).",
+)
+
+FAILED_GET_FILTER_ITEMS = LogMessage(
+ "E705",
+ "Failed to get filter items for history search (key: %(key)s).",
+)
+
+FAILED_GET_FILE_PATH = LogMessage(
+ "E706",
+ "Failed to get file path for file (id: %(file_id)s) from database.",
+)
UNNECESSARY_CONTRIB = LogMessage(
"E999", "Contrib utilities can only be used in development mode."
diff --git a/src/server/messages/info.py b/src/server/messages/info.py
index 8c5e825d..a65a805c 100644
--- a/src/server/messages/info.py
+++ b/src/server/messages/info.py
@@ -177,3 +177,35 @@
"I321",
"Successfully updated affiliations of User resource (id: %(id)s, ePPN: %(eppn)s).",
)
+
+
+SUCCESS_UPLOAD_FILES = LogMessage(
+ "I601",
+ "successfully uploaded file and create Files record: %(file_path)s",
+)
+
+SUCCESS_VALIDATE = LogMessage(
+ "I602",
+ "successfully validated the uploaded file: %(file_id)s",
+)
+
+SUCCESS_GET_VALIDATE_RESULT = LogMessage(
+ "I603",
+ "successfully retrieved validation result for history record: %(history_id)s",
+)
+
+SUCCESS_BULK_OPERATION = LogMessage(
+ "I604",
+ "successfully completed bulk operation for history record: %(history_id)s",
+)
+
+SUCCESS_GET_BULK_OPERATION_RESULT = LogMessage(
+ "I605",
+ "successfully retrieved bulk operation result for history record: %(history_id)s",
+)
+
+SUCCESS_UPDATE_PUBLIC_STATUS = LogMessage(
+ "I700",
+ "successfully update public status of history record (id: %(history_id)s) in"
+ " database.",
+)
diff --git a/src/server/services/bulks.py b/src/server/services/bulks.py
index 628ecddf..ee635c3d 100644
--- a/src/server/services/bulks.py
+++ b/src/server/services/bulks.py
@@ -13,7 +13,7 @@
from http import HTTPStatus
from itertools import zip_longest
from pathlib import Path
-from uuid import UUID
+from uuid import UUID, uuid7
import openpyxl
import requests
@@ -21,17 +21,19 @@
from celery import shared_task
from flask import current_app
from pydantic import ValidationError
+from redis.exceptions import ConnectionError as RedisConnectionError
from server.clients import bulks
from server.config import config
-from server.const import ValidationEntity
+from server.const import USER_ROLES, ValidationEntity
+from server.db import db
from server.entities.bulk import (
- CheckResult,
- HistorySummary,
+ EachResult,
+ ExecuteResults,
RepositoryMember,
ResultSummary,
UserAggregated,
- ValidateSummary,
+ ValidateResults,
)
from server.entities.bulk_request import BulkOperation
from server.entities.map_error import MapError
@@ -39,21 +41,70 @@
from server.entities.map_user import EPPN, Email, Group, MapUser
from server.entities.patch_request import RemoveOperation
from server.entities.summaries import GroupSummary
-from server.entities.user_detail import UserDetail
+from server.entities.user_detail import RepositoryRole, UserDetail
from server.exc import (
+ DatastoreError,
+ FileFormatError,
+ FileNotFound,
+ FileUploadError,
FileValidationError,
+ InvalidFormError,
OAuthTokenError,
RecordNotFound,
- ResourceInvalid,
- ResourceNotFound,
+ TaskExcutionError,
UnexpectedResponseError,
)
+from server.messages import E, I
+from server.services.utils.permissions import (
+ get_permitted_repository_ids,
+ is_current_user_system_admin,
+)
+from server.services.utils.transformers import validate_user_to_map_user
from . import groups, history_table, users, utils
from .token import get_access_token, get_client_secret
from .utils import session_required
+if t.TYPE_CHECKING:
+ from celery.result import AsyncResult
+ from werkzeug.datastructures import FileStorage
+
+
+def upload_file(repository_id: str, bulk_file: FileStorage) -> UUID:
+ """Upload a file for bulk processing.
+
+ Args:
+ repository_id (str): Target repository ID for upload.
+ bulk_file (FileStorage): File to upload.
+
+ Returns:
+ UUID: The ID of the temporary file.
+
+ Raises:
+ FileUploadError: If there is an error saving the uploaded file.
+ """
+ temp_id = uuid7()
+ temp_dir = Path(config.STORAGE.local.temporary)
+ original_filename = bulk_file.filename or "upload_file"
+ new_filename = f"{temp_id}_{Path(original_filename).name}"
+ file_path = temp_dir / new_filename
+ file_content = {"repositories": [{"id": repository_id}]}
+ history_table.create_file(
+ file_id=temp_id, file_path=str(file_path), file_content=file_content
+ )
+ try:
+ bulk_file.save(str(file_path))
+ except (PermissionError, FileNotFoundError) as exc:
+ db.session.rollback()
+ error = E.FAILED_SAVE_UPLOADED_FILE % {"file_path": file_path}
+ raise FileUploadError(error) from exc
+ db.session.commit()
+ current_app.logger.info(I.SUCCESS_UPLOAD_FILES, {"file_path": file_path})
+ delete_temporary_file.apply_async((str(temp_id),), countdown=3600)
+ return temp_id
+
+
@shared_task()
@session_required
def validate_upload_data(
@@ -76,8 +127,12 @@ def validate_upload_data(
data, new_data = build_user_from_file(file_path)
- updata_users: list[UserDetail] = build_user_detail_from_dict(data).root
- create_users: list[UserDetail] = build_user_detail_from_dict_by_name(new_data).root
+ updata_users: list[UserDetail] = build_user_detail_from_dict(
+ data, repository_id
+ ).root
+ create_users: list[UserDetail] = build_user_detail_from_dict_by_name(
+ new_data, repository_id
+ ).root
updata_users_id = {u.id for u in updata_users if u.id is not None}
missing_users = _get_missing_users(repository_member, updata_users_id)
repo_user_by_id = _get_repo_user_by_id(repository_member, updata_users_id)
@@ -86,15 +141,17 @@ def validate_upload_data(
updata_users, create_users, repository_member, repo_user_by_id
)
- results = ValidateSummary(
+ results = ValidateResults(
results=check_results, summary=summary, missing_user=missing_users
)
- return history_table.create_upload(
+ result = history_table.create_upload(
operator_id=operator_id,
operator_name=operator_name,
file_id=temp_file_id,
results=results.model_dump(mode="json"),
)
+ current_app.logger.info(I.SUCCESS_VALIDATE, {"file_id": temp_file_id})
+ return result.id
def get_repository_member(repository_id: str) -> RepositoryMember:
@@ -130,44 +187,46 @@ def build_user_from_file(
A tuple containing two dictionaries:
- The first dictionary contains user data keyed by user ID.
- The second dictionary contains new user data keyed by user name.
-
- Raises:
- ResourceNotFound: If the file does not exist.
"""
- try:
- gen = _read_file(file_path)
- except (ResourceInvalid, ResourceNotFound) as e:
- current_app.logger.error(e)
- raise ResourceNotFound(str(e)) from e
+ gen = _read_file(file_path)
it = next(gen)
- header = [("" if h is None else str(h).strip()) for h in next(it)]
- _ = next(it)
- id_idx = header.index("id")
- idx_of = {name: i for i, name in enumerate(header)}
data = defaultdict(lambda: defaultdict(list))
new_data = defaultdict(lambda: defaultdict(list))
- for row in it:
- if row is None:
+
+ idx_of = {}
+ id_idx = None
+ user_name_idx = None
+ header_row_index = 1
+
+ for i, row in enumerate(it):
+ if i == header_row_index + 1 or row is None:
continue
- r = list(row)
- rid = r[id_idx]
- if not rid:
+ if i == 0:
+ _ = (str(h).strip() if h is not None else "" for h in row)
+
+ if i == header_row_index:
+ header = [str(h).strip() if h is not None else "" for h in row]
+ idx_of = {name: idx for idx, name in enumerate(header)}
+ id_idx = idx_of.get("id")
user_name_idx = idx_of.get("user_name")
- if user_name_idx is None:
- continue
- user_name_value = r[user_name_idx]
- bucket = new_data[user_name_value]
- for col, j in idx_of.items():
- if col != "user_name":
- bucket[col].append(r[j])
continue
- bucket = data[rid]
+ r = list(row)
+ rid = r[id_idx] if id_idx is not None else None
+
+ if rid:
+ bucket = data[rid]
+ exclude = {"id"}
+ else:
+ user_name_value = r[user_name_idx] if user_name_idx else None
+ bucket = new_data[user_name_value]
+ exclude = {"user_name"}
for col, j in idx_of.items():
- if col != "id":
+ if col not in exclude:
bucket[col].append(r[j])
+
return dict(data), dict(new_data)
@@ -177,15 +236,14 @@ def _read_file(file_path: str) -> t.Generator:
Args:
file_path (str): Path to the input file containing user data.
-
Raises:
- ResourceNotFound: If the file does not exist.
- ResourceInvalid : If the format is unsupported or parsing fails.
+ FileNotFound: If the file does not exist.
+ FileFormatError : If the format is unsupported or parsing fails.
"""
path = Path(file_path)
if not path.exists():
- error = f"{path}: File not found."
- raise ResourceNotFound(error)
+ error = E.FILE_EXPIRED % {"path": path}
+ raise FileNotFound(error)
iterator = None
suffix = path.suffix.lower()
@@ -200,14 +258,14 @@ def _read_file(file_path: str) -> t.Generator:
if ws:
iterator = ws.iter_rows(values_only=True)
if iterator is None:
- error = f"{path.suffix}: Unsupported file format."
- raise ResourceInvalid(error)
+ error = E.FILE_FORMAT_UNSUPPORTED % {"suffix": path.suffix}
+ raise FileFormatError(error)
yield iterator
def build_user_detail_from_dict(
- data: dict[str, dict[str, list[str]]],
+ data: dict[str, dict[str, list[str]]], repository_id: str
) -> UserAggregated:
"""Build UserAggregated from a dictionary of user data.
@@ -215,10 +273,14 @@ def build_user_detail_from_dict(
data (dict[str, dict[str, list[str]]]):
A dictionary where the key is the user ID and the value is another
dictionary containing user attributes.
+ repository_id (str): The ID of the repository to which the users belong.
Returns:
UserAggregated:
The aggregated user details built from the input dictionary.
+
+ Raises:
+ FileValidationError: If there is an error validating the user data.
"""
users: list[UserDetail] = []
@@ -237,6 +299,11 @@ def build_user_detail_from_dict(
if preferred_language_list is not None and len(preferred_language_list) > 0
else ""
)
+ str_role = columns.get("role")
+ role = str_role[0] if str_role and len(str_role) > 0 else None
+ repository_roles: list[RepositoryRole] = [
+ RepositoryRole(id=repository_id, user_role=_resolve_non_sysadmin_role(role))
+ ]
eppns: set[str] = set(columns.get("edu_person_principal_names[]") or [])
emails: set[str] = set(columns.get("emails[]") or [])
@@ -263,15 +330,19 @@ def build_user_detail_from_dict(
user_name=user_name,
emails=emails,
preferred_language=preferred_language,
+ repository_roles=repository_roles,
groups=groups,
)
)
-
- return UserAggregated(root=users)
+ try:
+ return UserAggregated(root=users)
+ except ValidationError as exc:
+ current_app.logger.error(exc)
+ raise FileValidationError(E.INVALID_FILE_STRUCTURE) from exc
def build_user_detail_from_dict_by_name(
- data: dict[str, dict[str, list[str]]],
+ data: dict[str, dict[str, list[str]]], repository_id: str
) -> UserAggregated:
"""Build UserAggregated from a dictionary of user data.
@@ -279,10 +350,14 @@ def build_user_detail_from_dict_by_name(
data (dict[str, dict[str, list[str]]]):
A dictionary where the key is the user name and the value is another
dictionary containing user attributes.
+ repository_id (str): The ID of the repository to which the users belong.
Returns:
UserAggregated:
The aggregated user details built from the input dictionary.
+
+ Raises:
+ FileValidationError: If there is an error validating the user data.
"""
users: list[UserDetail] = []
@@ -303,6 +378,11 @@ def build_user_detail_from_dict_by_name(
gid_list: list[str] = list(columns.get("groups[].id") or [])
gname_list: list[str] = list(columns.get("groups[].name") or [])
+ str_role = columns.get("role")
+ role = str_role[0] if str_role and len(str_role) > 0 else None
+ repository_roles: list[RepositoryRole] = [
+ RepositoryRole(id=repository_id, user_role=_resolve_non_sysadmin_role(role))
+ ]
groups: list[GroupSummary] = []
seen = set()
for gid, gname in zip_longest(gid_list, gname_list, fillvalue=None):
@@ -323,11 +403,22 @@ def build_user_detail_from_dict_by_name(
user_name=user_name,
emails=emails,
preferred_language=preferred_language,
+ repository_roles=repository_roles,
groups=groups,
)
)
+ try:
+ return UserAggregated(root=users)
+ except ValidationError as exc:
+ current_app.logger.error(exc)
+ raise FileValidationError(E.INVALID_FILE_STRUCTURE) from exc
- return UserAggregated(root=users)
+
+_ROLE_MAP = {m.value: m for m in USER_ROLES if m is not USER_ROLES.SYSTEM_ADMIN}
+
+
+def _resolve_non_sysadmin_role(role: str | None) -> USER_ROLES | None:
+ return _ROLE_MAP.get(role) if role else None
def _get_missing_users(
@@ -363,13 +454,13 @@ def _build_check_results(
create_users: list[UserDetail],
repository_member: RepositoryMember,
repo_user_by_id: dict[str, UserDetail],
-) -> tuple[list[CheckResult], HistorySummary]:
+) -> tuple[list[EachResult], ResultSummary]:
count_create = 0
count_update = 0
count_delete = 0
count_skip = 0
count_error = 0
- check_results: list[CheckResult] = []
+ check_results: list[EachResult] = []
for u in create_users:
code = None
user_group_ids = {g.id for g in u.groups} if u.groups else set()
@@ -377,7 +468,7 @@ def _build_check_results(
if not user_group_ids.issubset(repository_member.groups):
code = "Group ID does not exist"
check_results.append(
- CheckResult(
+ EachResult(
id=u.id,
eppn=u.eppns or [],
user_name=u.user_name,
@@ -389,25 +480,25 @@ def _build_check_results(
)
count_error += 1
continue
-
- if not _check_value(u):
- code = "Invalid user data"
+ try:
+ validate_user_to_map_user(u, mode="create")
+ except InvalidFormError as exc:
check_results.append(
- CheckResult(
+ EachResult(
id=u.id,
eppn=u.eppns or [],
user_name=u.user_name,
groups=user_group_ids,
email=u.emails or [],
status="error",
- code=code,
+ code=exc.message,
)
)
count_error += 1
continue
check_results.append(
- CheckResult(
+ EachResult(
id=u.id,
eppn=u.eppns or [],
user_name=u.user_name,
@@ -436,7 +527,7 @@ def _build_check_results(
count_skip += 1
code = _check_immutable_attributes(repo_user, u)
check_results.append(
- CheckResult(
+ EachResult(
id=u.id,
eppn=repo_user.eppns or [],
user_name=repo_user.user_name,
@@ -446,7 +537,7 @@ def _build_check_results(
code=code,
)
)
- summary = HistorySummary(
+ summary = ResultSummary(
create=count_create,
update=count_update,
delete=count_delete,
@@ -467,7 +558,7 @@ def _check_value(user: UserDetail) -> bool:
"""
if not user.id:
return False
- if not re.compile(r"^[A-Za-z0-9]{1,50}$").fullmatch(user.id):
+ if not re.compile(r"^[A-Za-z0-9._-]{1,50}$").fullmatch(user.id):
return False
return len(user.user_name) <= ValidationEntity.USER_NAME_MAX_LENGTH
@@ -484,22 +575,46 @@ def _check_immutable_attributes(
Returns:
str | None: The name of the immutable attribute if found, None otherwise.
"""
+ if original.user_name != update_user.user_name:
+ return "user_name is immutable"
if original.emails != update_user.emails:
- return "emails"
+ return "emails are immutable"
if original.eppns != update_user.eppns:
- return "eppns"
+ return "eppns are immutable"
if original.preferred_language != update_user.preferred_language:
- return "preferred_language"
- if original.last_modified != update_user.last_modified:
- return "last_modified"
- if original.created != update_user.created:
- return "created"
+ return "preferred_language is immutable"
return None
+def get_validate_task_result(task_id: str) -> AsyncResult[UUID]:
+ """Get the result of a validation task.
+
+ Args:
+ task_id (str): The ID of the validation task.
+
+ Returns:
+ AsyncResult[UUID]: The result of the validation task.
+
+ Raises:
+ DatastoreError: If there is an error connecting to the datastore.
+ TaskExcutionError: If the task with the given ID does not exist.
+ """
+ try:
+ res = validate_upload_data.AsyncResult(task_id)
+ except RedisConnectionError as exc:
+ error = E.FAILED_CONNECT_REDIS % {"error": str(exc)}
+ current_app.logger.error(error)
+ raise DatastoreError(error) from exc
+ if not res:
+ error = E.TASK_NOT_FOUND % {"task_id": task_id}
+ current_app.logger.error(error)
+ raise TaskExcutionError(error)
+ return res
+
+
def get_validate_result(
history_id: UUID, status_filter: list[str], offset: int, size: int
-) -> ValidateSummary:
+) -> ValidateResults:
"""Get the validation result summary for the specified upload history ID.
Args:
@@ -514,51 +629,53 @@ def get_validate_result(
results = history_table.get_paginated_upload_results(
history_id, offset, size, status_filter
)
- check_results = [CheckResult.model_validate(it) for it in results]
+ check_results = [EachResult.model_validate(it) for it in results]
summary = history_table.get_upload_results(history_id, "summary")
missing_user = history_table.get_upload_results(history_id, "missingUser")
- return ValidateSummary.model_validate({
+ result = ValidateResults.model_validate({
"results": check_results,
"summary": summary,
"missingUser": missing_user or [],
"offset": offset,
"pageSize": size,
})
+ current_app.logger.info(I.SUCCESS_GET_VALIDATE_RESULT, {"history_id": history_id})
+ return result
@shared_task()
def update_users(
- history_id: UUID, temp_file_id: UUID, delete_users: list[str] | None
+ history_id: UUID, temp_file_id: UUID, remove_users: list[str] | None
) -> UUID:
"""Perform bulk update of users based on the validation results.
Args:
history_id (UUID): The ID of the upload history.
temp_file_id (UUID): The ID of the temporary file.
- delete_users (list[str] | None): The list of user IDs to be deleted.
+ remove_users (list[str] | None):
+ The list of user IDs to be removed by repository.
Returns:
UUID: The ID of the upload history record.
Raises:
- ResourceNotFound: If the upload history does not exist.
+ FileNotFound: If the upload history does not exist.
requests.RequestException: If there is an error communicating with mAP Core API.
ValidationError: If there is an error parsing the response from mAP Core API.
OAuthTokenError: If there is an issue with the access token.
UnexpectedResponseError: If there is an unexpected response from mAP Core API.
- ResourceInvalid: If there is an invalid resource error from mAP Core API.
FileValidationError: If there are errors in the validation results.
"""
upload_data = history_table.get_upload_by_id(history_id)
if not upload_data:
error = f"History not found: {history_id}"
current_app.logger.error(error)
- raise ResourceNotFound(error)
+ raise FileNotFound(error)
# file content must contain at least one repository.
repository_id = upload_data.file.file_content["repositories"][0]["id"]
- check_results: list[CheckResult] = upload_data.results.get("results", [])
+ check_results: list[EachResult] = upload_data.results.get("results", [])
summary = upload_data.results.get("summary", {})
if summary.get("error", 1) > 0:
@@ -567,7 +684,7 @@ def update_users(
raise FileValidationError(error)
bulk_ops, count_delete = _build_bulk_operations_from_check_results(
- repository_id, check_results, delete_users
+ repository_id, check_results, remove_users
)
summary.update({"delete": count_delete})
@@ -597,8 +714,9 @@ def update_users(
raise
if isinstance(result, MapError):
- current_app.logger.info(result.detail)
- raise ResourceInvalid(result.detail)
+ current_app.logger.error(E.RECEIVE_RESPONSE_MESSAGE, {"message": result.detail})
+ error = E.FAILED_BULK_OPERATION
+ raise UnexpectedResponseError(result.detail)
count_error = 0
for i, operation in enumerate(result.operations):
@@ -623,6 +741,7 @@ def update_users(
status="S",
)
+ current_app.logger.info(I.SUCCESS_BULK_OPERATION, {"history_id": history_id})
return history_id
@@ -636,25 +755,25 @@ def save_file(temp_file_id: UUID) -> UUID:
UUID: The ID of the saved permanent file.
Raises:
- ResourceNotFound: If the temporary file does not exist.
- ResourceInvalid: If the file format is invalid.
+ FileNotFound: If the temporary file does not exist.
+ FileFormatError: If the file format is invalid.
"""
try:
files = history_table.get_file_by_id(temp_file_id)
repository_id = files.file_content["repositories"][0]["id"]
except (KeyError, AttributeError) as e:
current_app.logger.error("Failed to retrieve temporary file: %s", temp_file_id)
- raise ResourceNotFound(str(e)) from e
+ raise FileNotFound(str(e)) from e
file_path = Path(files.file_path)
if file_path.parent != Path(config.STORAGE.local.temporary):
return files.id
if not file_path.exists():
error_msg = f"File not found: {file_path}"
- raise ResourceNotFound(error_msg)
+ raise FileNotFound(error_msg)
if file_path.suffix not in {".csv", ".tsv", ".xlsx"}:
- error = "not supported file format."
- raise ResourceInvalid(error)
+ error = E.FILE_FORMAT_UNSUPPORTED % {"suffix": file_path.suffix}
+ raise FileFormatError(error)
target_dir = Path(config.STORAGE.local.storage) / datetime.now(UTC).strftime(
"%Y/%m"
@@ -665,10 +784,10 @@ def save_file(temp_file_id: UUID) -> UUID:
return history_table.create_file(
file_path=str(target_path),
file_content={"repositories": [{"id": repository_id}]},
- )
+ ).id
-def build_map_user_from_check_result(user: CheckResult) -> MapUser:
+def build_map_user_from_check_result(user: EachResult) -> MapUser:
"""Build MapUser from CheckResult.
Args:
@@ -718,7 +837,7 @@ def build_remove_user_path(user: UserDetail, repository_id: str) -> BulkOperatio
def _build_bulk_operations_from_check_results(
- repository_id: str, check_results: list[CheckResult], delete_users: list[str] | None
+ repository_id: str, check_results: list[EachResult], remove_users: list[str] | None
) -> tuple[list[BulkOperation], int]:
repository_member = get_repository_member(repository_id)
@@ -761,7 +880,7 @@ def _build_bulk_operations_from_check_results(
])
delete_user_list = users.search(
- utils.make_criteria_object("users", i=delete_users), raw=True
+ utils.make_criteria_object("users", i=remove_users), raw=True
).resources
group_user_ops = {}
@@ -776,7 +895,7 @@ def _build_bulk_operations_from_check_results(
"remove"
].add(user.id)
check_results.append(
- CheckResult(
+ EachResult(
id=user.id,
eppn=user.eppns or [],
user_name=user.user_name,
@@ -822,9 +941,35 @@ def _build_groups_update_bulk_operations(
return bulk_ops
+def get_execute_task_result(task_id: str) -> AsyncResult[UUID]:
+ """Get the result of an execution task.
+
+ Args:
+ task_id (str): The ID of the execution task.
+
+ Returns:
+ AsyncResult[UUID]: The result of the execution task.
+
+ Raises:
+ DatastoreError: If there is an error connecting to the datastore.
+ TaskExcutionError: If the task with the given ID does not exist.
+ """
+ try:
+ res = update_users.AsyncResult(task_id)
+ except RedisConnectionError as exc:
+ error = E.FAILED_CONNECT_REDIS % {"error": str(exc)}
+ current_app.logger.error(error)
+ raise DatastoreError(error) from exc
+ if not res:
+ error = E.TASK_NOT_FOUND % {"task_id": task_id}
+ current_app.logger.error(error)
+ raise TaskExcutionError(error)
+ return res
+
+
def get_upload_result(
history_id: UUID, status_filter: list[str], offset: int, size: int
-) -> ResultSummary:
+) -> ExecuteResults:
"""Get the bulk operation result summary with filtering and pagination.
Args:
@@ -834,14 +979,14 @@ def get_upload_result(
offset (int): The offset for pagination.
Returns:
- ResultSummary: The summary of the bulk operation result.
+ ExecuteResults: The summary of the bulk operation result.
Raises:
RecordNotFound: If the upload history with the given ID does not exist.
"""
upload = history_table.get_upload_by_id(history_id)
if not upload:
- error = f"upload history not found: {history_id}"
+ error = E.UPDATE_HISTORY_RECORD_NOT_FOUND % {"id": history_id}
raise RecordNotFound(error)
raw_results: list[dict] = history_table.get_paginated_upload_results(
@@ -861,7 +1006,11 @@ def get_upload_result(
"offset": offset,
"pageSize": size,
}
- return ResultSummary.model_validate(payload)
+ result = ExecuteResults.model_validate(payload)
+ current_app.logger.info(
+ I.SUCCESS_GET_BULK_OPERATION_RESULT, {"history_id": history_id}
+ )
+ return result
@shared_task()
@@ -879,3 +1028,47 @@ def delete_temporary_file(temp_id: str) -> None:
if Path(file_path).exists():
Path(file_path).unlink()
history_table.delete_file_by_id(temp_file_id)
+
+
+def chack_permission_to_operation(history_id: UUID, operator_id: str) -> bool:
+ """Check if the user has permission to perform bulk operation.
+
+ Args:
+ history_id (UUID): The ID of the upload history.
+ operator_id (str): The ID of the operator.
+
+ Returns:
+ bool: True if the user has permission, False otherwise.
+
+ Raises:
+ RecordNotFound: If the upload history with the given ID does not exist.
+ """
+ upload = history_table.get_upload_by_id(history_id)
+ if not upload:
+ error = E.UPDATE_HISTORY_RECORD_NOT_FOUND % {"id": history_id}
+ current_app.logger.error(error)
+ raise RecordNotFound(error)
+ return upload.operator_id == operator_id
+
+
+def chack_permission_to_view(history_id: UUID) -> bool:
+ """Check if the user has permission to view the specified upload history.
+
+ Args:
+ history_id (UUID): The ID of the upload history.
+
+ Returns:
+ bool: True if the user has permission, False otherwise.
+
+ Raises:
+ RecordNotFound: If the upload history with the given ID does not exist.
+ """
+ if is_current_user_system_admin():
+ return True
+ upload = history_table.get_upload_by_id(history_id)
+ if not upload:
+ error = E.UPDATE_HISTORY_RECORD_NOT_FOUND % {"id": history_id}
+ current_app.logger.error(error)
+ raise RecordNotFound(error)
+ repository_id = upload.file.file_content["repositories"][0]["id"]
+ return repository_id not in get_permitted_repository_ids() and upload.public
diff --git a/src/server/services/history.py b/src/server/services/history.py
index bc3789e6..5bdf046f 100644
--- a/src/server/services/history.py
+++ b/src/server/services/history.py
@@ -7,6 +7,7 @@
import typing as t
from datetime import date, datetime, time, timedelta
+from pathlib import Path
from types import SimpleNamespace
from uuid import UUID
@@ -14,6 +15,7 @@
from sqlalchemy import cast, func, or_, select
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.exc import SQLAlchemyError
+from sqlalchemy.orm import aliased
from server.const import DEFAULT_SEARCH_COUNT
from server.db.history import DownloadHistory, Files, UploadHistory
@@ -25,6 +27,7 @@
from server.entities.search_request import SearchResult
from server.entities.summaries import UserSummary
from server.exc import DatabaseError, InvalidQueryError, RecordNotFound
+from server.messages import E, I
from .utils import (
get_permitted_repository_ids,
@@ -49,25 +52,31 @@ def get_upload_history_data(
Returns:
SearchResult: upload history data with pagination.
+ Raises:
+ DatabaseError: If a database operation fails.
"""
filters = _build_filters_for_history(criteria, history_type="upload")
base_stmt = select(UploadHistory, Files).join(UploadHistory.file).filter(*filters)
count_stmt = select(func.count()).select_from(base_stmt.subquery())
+ try:
+ total = db.session.execute(count_stmt).scalar_one()
- total = db.session.execute(count_stmt).scalar_one()
-
- if criteria.d == "desc":
- stmt = base_stmt.order_by(UploadHistory.timestamp.desc())
- else:
- stmt = base_stmt.order_by(UploadHistory.timestamp.asc())
+ if criteria.d == "desc":
+ stmt = base_stmt.order_by(UploadHistory.timestamp.desc())
+ else:
+ stmt = base_stmt.order_by(UploadHistory.timestamp.asc())
- page = criteria.p or 1
- page_size = criteria.l or DEFAULT_SEARCH_COUNT
- offset = (page - 1) * page_size
- stmt = stmt.limit(page_size).offset(offset)
+ page = criteria.p or 1
+ page_size = criteria.l or DEFAULT_SEARCH_COUNT
+ offset = (page - 1) * page_size
+ stmt = stmt.limit(page_size).offset(offset)
- rows = db.session.execute(stmt).all()
+ rows = db.session.execute(stmt).all()
+ except SQLAlchemyError as exc:
+ current_app.logger.error(str(exc))
+ error = E.FAILED_GET_HISTORY_RECORDS % {"table": "upload"}
+ raise DatabaseError(error) from exc
data = [
{
**history.__dict__,
@@ -174,6 +183,8 @@ def get_download_history_data(
Returns:
SearchResult: download history data with pagination.
+ Raises:
+ DatabaseError: If a database operation fails.
"""
filters = _build_filters_for_history(criteria, history_type="download")
@@ -181,26 +192,33 @@ def get_download_history_data(
select(DownloadHistory, Files).join(DownloadHistory.file).filter(*filters)
)
count_stmt = select(func.count()).select_from(base_stmt.subquery())
- total = db.session.execute(count_stmt).scalar_one()
-
- subq = (
- select(func.count(DownloadHistory.id))
- .where(DownloadHistory.parent_id == DownloadHistory.id)
- .scalar_subquery()
- )
- stmt = base_stmt.add_columns(subq)
+ try:
+ total = db.session.execute(count_stmt).scalar_one()
+
+ child = aliased(DownloadHistory)
+ subq = (
+ select(func.count(child.id))
+ .where(child.parent_id == DownloadHistory.id)
+ .correlate(DownloadHistory)
+ .scalar_subquery()
+ )
+ stmt = base_stmt.add_columns(subq)
- if criteria.d == "desc":
- stmt = stmt.order_by(DownloadHistory.timestamp.desc())
- else:
- stmt = stmt.order_by(DownloadHistory.timestamp.asc())
+ if criteria.d == "desc":
+ stmt = stmt.order_by(DownloadHistory.timestamp.desc())
+ else:
+ stmt = stmt.order_by(DownloadHistory.timestamp.asc())
- page = criteria.p or 1
- page_size = criteria.l or DEFAULT_SEARCH_COUNT
- offset = (page - 1) * page_size
- stmt = stmt.limit(page_size).offset(offset)
+ page = criteria.p or 1
+ page_size = criteria.l or DEFAULT_SEARCH_COUNT
+ offset = (page - 1) * page_size
+ stmt = stmt.limit(page_size).offset(offset)
- rows = db.session.execute(stmt).all()
+ rows = db.session.execute(stmt).all()
+ except SQLAlchemyError as exc:
+ current_app.logger.error(str(exc))
+ error = E.FAILED_GET_HISTORY_RECORDS % {"table": "download"}
+ raise DatabaseError(error) from exc
data = [
{
**history.__dict__,
@@ -211,6 +229,7 @@ def get_download_history_data(
"group_count": len(file.file_content.get("groups", [])),
"user_count": len(file.file_content.get("users", [])),
"children_count": children_count,
+ "file_exists": Path(file.file_path).exists(),
}
for history, file, children_count in t.cast(
"t.Sequence[tuple[DownloadHistory, Files, int]]", rows
@@ -219,7 +238,7 @@ def get_download_history_data(
return SearchResult[DownloadHistoryData].model_validate(
{
"total": total,
- "pageSize": page_size,
+ "page_size": page_size,
"offset": offset,
"resources": data,
},
@@ -242,6 +261,7 @@ def get_filter_items(
Raises:
InvalidQueryError: If the filter key is invalid.
+ DatabaseError: If a database operation fails.
"""
table = UploadHistory if tab == "upload" else DownloadHistory
if key == "o":
@@ -250,8 +270,12 @@ def get_filter_items(
page_size = criteria.l or DEFAULT_SEARCH_COUNT
offset = (page - 1) * page_size
stmt = stmt.limit(page_size).offset(offset)
-
- results = db.session.execute(stmt).all()
+ try:
+ results = db.session.execute(stmt).all()
+ except SQLAlchemyError as exc:
+ current_app.logger.error(str(exc))
+ error = E.FAILED_GET_HISTORY_RECORDS % {"table": tab}
+ raise DatabaseError(error) from exc
items = [
UserSummary(id=operator_id, user_name=operator_name)
for operator_id, operator_name in results
@@ -260,7 +284,7 @@ def get_filter_items(
resources=items, total=0, page_size=page_size, offset=offset
)
- error = f"Unsupported criteria type: {type(criteria)}"
+ error = E.FAILED_GET_FILTER_ITEMS % {"key": key}
current_app.logger.error(error)
raise InvalidQueryError(error)
@@ -287,14 +311,19 @@ def update_public_status(
record = db.session.get(table, history_id)
if record is None:
- error = f"{history_id} is not found"
+ error = E.FAILED_GET_HISTORY_RECORD % {
+ "history_id": history_id,
+ "table": tab,
+ }
raise RecordNotFound(error)
record.public = public
except SQLAlchemyError as exc:
- error = "Failed to update the public status due to a database error."
+ current_app.logger.error(str(exc))
+ error = E.FAILED_UPDATE_PUBLIC % {"history_id": history_id}
raise DatabaseError(error) from exc
db.session.commit()
+ current_app.logger.info(I.SUCCESS_UPDATE_PUBLIC_STATUS)
return record.public
@@ -314,11 +343,12 @@ def get_file_path(file_id: UUID) -> str:
try:
file = db.session.get(Files, file_id)
except SQLAlchemyError as exc:
- error = "Failed to retrieve the file path due to a database error."
+ current_app.logger.error(str(exc))
+ error = E.FAILED_GET_FILE_PATH % {"file_id": file_id}
raise DatabaseError(error) from exc
if not file:
- error = f"File with ID {file_id} not found."
+ error = E.FAILED_GET_FILE_PATH % {"file_id": file_id}
raise RecordNotFound(error)
return file.file_path
diff --git a/src/server/services/history_table.py b/src/server/services/history_table.py
index ff97c18e..ec43a1b7 100644
--- a/src/server/services/history_table.py
+++ b/src/server/services/history_table.py
@@ -11,11 +11,18 @@
from flask import current_app
from sqlalchemy import func
+from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import selectinload
from server.db import db
-from server.db.history import Files, UploadHistory
-from server.exc import InvalidQueryError, InvalidRecordError, RecordNotFound
+from server.db.history import DownloadHistory, Files, UploadHistory
+from server.exc import (
+ DatabaseError,
+ InvalidQueryError,
+ InvalidRecordError,
+ RecordNotFound,
+)
+from server.messages import E
def get_upload_by_id(history_id: UUID) -> UploadHistory | None:
@@ -26,10 +33,19 @@ def get_upload_by_id(history_id: UUID) -> UploadHistory | None:
Returns:
UploadHistory | None: The upload history record, or None if not found.
+
+ Raises:
+ DatabaseError: If there is an error querying the database.
"""
- return db.session.get(
- UploadHistory, history_id, options=[selectinload(UploadHistory.file)]
- )
+ try:
+ history = db.session.get(
+ UploadHistory, history_id, options=[selectinload(UploadHistory.file)]
+ )
+ except SQLAlchemyError as exc:
+ current_app.logger.error(str(exc))
+ error = E.FAILED_GET_UPLOAD_HISTORY_RECORD % {"history_id": history_id}
+ raise DatabaseError(error) from exc
+ return history
def get_upload_results(history_id: UUID, attribute: str) -> dict:
@@ -41,14 +57,27 @@ def get_upload_results(history_id: UUID, attribute: str) -> dict:
Returns:
dict: The upload results for the specified attribute.
+
+ Raises:
+ DatabaseError: If there is an error querying the database.
+ RecordNotFound: If no history record is found for the given ID.
"""
- result = (
- db.session
- .query(UploadHistory.results[attribute])
- .filter(UploadHistory.id == history_id)
- .first()
- )
- return result[0] if result else {}
+ try:
+ result = (
+ db.session
+ .query(UploadHistory.results[attribute])
+ .filter(UploadHistory.id == history_id)
+ .first()
+ )
+ except SQLAlchemyError as exc:
+ current_app.logger.error(str(exc))
+ error = E.FAILED_GET_UPLOAD_HISTORY_RECORD % {"history_id": history_id}
+ raise DatabaseError(error) from exc
+ if result is None:
+ error = E.FAILED_GET_FILE_RECORD % {"file_id": history_id}
+ current_app.logger.error(error)
+ raise RecordNotFound(error)
+ return result[0]
def get_paginated_upload_results(
@@ -67,34 +96,40 @@ def get_paginated_upload_results(
Raises:
InvalidQueryError: If offset or size is less than 1.
+ DatabaseError: If there is an error querying the database.
"""
if offset < 1 or size < 1:
- error_message = "Invalid offset or size"
- raise InvalidQueryError(error_message)
+ error = E.INVALID_Query % {"offset": offset, "size": size}
+ current_app.logger.error(error)
+ raise InvalidQueryError(error)
elements = func.jsonb_array_elements(
UploadHistory.results["results"]
).column_valued("item")
- query = (
- db.session
- .query(elements)
- .select_from(UploadHistory)
- .filter(UploadHistory.id == history_id)
- )
-
- if status_filter:
- query = query.filter(elements.op("->>")("status").in_(status_filter))
-
- offset_val = (offset - 1) * size
-
- raw_results = query.limit(size).offset(offset_val).all()
-
+ try:
+ query = (
+ db.session
+ .query(elements)
+ .select_from(UploadHistory)
+ .filter(UploadHistory.id == history_id)
+ )
+
+ if status_filter:
+ query = query.filter(elements.op("->>")("status").in_(status_filter))
+
+ offset_val = (offset - 1) * size
+
+ raw_results = query.limit(size).offset(offset_val).all()
+ except SQLAlchemyError as exc:
+ current_app.logger.error(str(exc))
+ error = E.FAILED_GET_UPLOAD_HISTORY_RECORD % {"history_id": history_id}
+ raise DatabaseError(error) from exc
return [r[0] for r in raw_results]
def create_upload(
file_id: UUID, results: dict, operator_id: str, operator_name: str
-) -> UUID:
+) -> UploadHistory:
"""Create a new upload history record.
Args:
@@ -104,30 +139,35 @@ def create_upload(
operator_name (str): The name of the operator performing the upload.
Returns:
- UUID: The ID of the newly created upload history record.
+ UploadHistory: The newly created upload history record.
Raises:
InvalidRecordError: If the results dictionary is missing required keys.
+ DatabaseError:
+ If there is an error creating the upload history record in the database.
"""
- history_record = UploadHistory()
- history_record.file_id = file_id
summary = results.get("summary")
items = results.get("results")
missing_users = results.get("missing_users", [])
if summary is None or items is None:
- error_message = "Results must include 'summary' and 'results' keys"
- raise InvalidRecordError(error_message)
-
- history_record.results = {
- "summary": summary,
- "items": items,
- "missing_users": missing_users,
- }
- history_record.operator_id = operator_id
- history_record.operator_name = operator_name
- db.session.add(history_record)
- db.session.commit()
- return history_record.id
+ raise InvalidRecordError(E.INVALID_UPLOAD_HISTORY_RECORD_ATTRIBUTES)
+ try:
+ history_record = UploadHistory()
+ history_record.file_id = file_id
+ history_record.results = {
+ "summary": summary,
+ "items": items,
+ "missing_users": missing_users,
+ }
+ history_record.operator_id = operator_id
+ history_record.operator_name = operator_name
+ db.session.add(history_record)
+ db.session.commit()
+ except SQLAlchemyError as exc:
+ current_app.logger.error(str(exc))
+ error = E.FAILED_CREATE_UPLOAD_HISTORY_RECORD % {"file_id": file_id}
+ raise DatabaseError(error) from exc
+ return history_record
def update_upload_status(
@@ -138,34 +178,42 @@ def update_upload_status(
) -> None:
"""Update the status of an upload history record.
+ Must call db.session.commit() after using this function to persist changes.
+
Args:
history_id (UUID): The ID of the history record to update.
status (Literal["P", "S", "F"]):
The new status ("P": Progress, "S": Success, "F": Failed).
new_results (dict | None): New results to update, if any.
file_id (UUID | None): New file ID to update, if any.
- """
- obj = db.session.get(UploadHistory, history_id)
- if obj is None:
- return
- if new_results:
- obj.results = {
- "summary": new_results.get("summary", {}),
- "items": new_results.get("results", []),
- "missing_users": new_results.get("missing_users", []),
- }
-
- obj.status = status
- now = datetime.now(UTC)
- if status == "P":
- obj.timestamp = now
- else:
- obj.end_timestamp = now
-
- if file_id:
- obj.file_id = file_id
- db.session.commit()
+ Raises:
+ DatabaseError: If there is an error updating the database.
+ """
+ try:
+ obj = db.session.get(UploadHistory, history_id)
+ if obj is None:
+ return
+ if new_results:
+ obj.results = {
+ "summary": new_results.get("summary", {}),
+ "items": new_results.get("results", []),
+ "missing_users": new_results.get("missing_users", []),
+ }
+
+ obj.status = status
+ now = datetime.now(UTC)
+ if status == "P":
+ obj.timestamp = now
+ else:
+ obj.end_timestamp = now
+
+ if file_id:
+ obj.file_id = file_id
+ except SQLAlchemyError as exc:
+ current_app.logger.error(str(exc))
+ error = E.FAILED_UPDATE_HISTORY_RECORD_STATUS % {"history_id": history_id}
+ raise DatabaseError(error) from exc
def get_history_by_file_id(file_id: UUID) -> UploadHistory:
@@ -179,8 +227,16 @@ def get_history_by_file_id(file_id: UUID) -> UploadHistory:
Raises:
RecordNotFound: If no history record is found for the file ID.
+ DatabaseError: If there is an error querying the database.
"""
- result = db.session.query(UploadHistory).filter_by(file_id=file_id).one_or_none()
+ try:
+ result = (
+ db.session.query(UploadHistory).filter_by(file_id=file_id).one_or_none()
+ )
+ except SQLAlchemyError as exc:
+ current_app.logger.error(str(exc))
+ error = E.FAILED_GET_UPLOAD_HISTORY_RECORD_BY_FILE_ID % {"file_id": file_id}
+ raise DatabaseError(error) from exc
if result is None:
error = f"History not found for file_id: {file_id}"
current_app.logger.error(error)
@@ -199,8 +255,14 @@ def get_file_by_id(file_id: UUID) -> Files:
Raises:
RecordNotFound: If no file record is found for the file ID.
+ DatabaseError: If there is an error querying the database.
"""
- result = db.session.query(Files).filter_by(id=file_id).one_or_none()
+ try:
+ result = db.session.query(Files).filter_by(id=file_id).one_or_none()
+ except SQLAlchemyError as exc:
+ current_app.logger.error(str(exc))
+ error = E.FAILED_GET_FILE_RECORD % {"file_id": file_id}
+ raise DatabaseError(error) from exc
if result is None:
error = f"File not found for file_id: {file_id}"
current_app.logger.error(error)
@@ -211,35 +273,92 @@ def get_file_by_id(file_id: UUID) -> Files:
def delete_file_by_id(file_id: UUID) -> None:
"""Delete a file record by its ID.
+ Must call db.session.commit() after using this function to persist changes.
+
Args:
file_id (UUID): The ID of the file to delete.
+
+ Raises:
+ DatabaseError: If there is an error deleting the file from the database.
"""
- Files.query.filter(Files.id == file_id).delete()
- db.session.commit()
+ try:
+ Files.query.filter(Files.id == file_id).delete()
+ except SQLAlchemyError as exc:
+ current_app.logger.error(str(exc))
+ error = E.FAILED_DELETE_FILE_RECORD % {"file_id": file_id}
+ raise DatabaseError(error) from exc
def create_file(
file_path: str, file_content: dict, file_id: UUID | None = None
-) -> UUID:
+) -> Files:
"""Create or update a file record.
+ Must call db.session.commit() after using this function to persist changes.
+
Args:
file_path (str): The path of the file.
file_content (dict): The content of the file.
file_id (UUID | None): The ID of the file to update.
Returns:
- UUID: The ID of the created or updated file.
+ Files: The created or updated file record.
+
+ Raises:
+ DatabaseError: If there is an error creating the file record in the database.
+ """
+ try:
+ file_record = Files()
+ if file_id:
+ file_record.id = file_id
+ file_record.file_path = str(file_path)
+ file_record.file_content = {
+ "repositories": file_content.get("repositories", []),
+ "groups": file_content.get("groups", []),
+ "users": file_content.get("users", []),
+ }
+ db.session.add(file_record)
+ except SQLAlchemyError as exc:
+ current_app.logger.error(str(exc))
+ error = E.FAILED_CREATE_FILE_RECORD % {"file_path": file_path}
+ raise DatabaseError(error) from exc
+ return file_record
+
+
+def create_download_history(
+ file_id: UUID,
+ file_path: str,
+ file_content: dict,
+ operator_id: str,
+ operator_name: str,
+) -> DownloadHistory:
+ """Create a new download history record.
+
+ Must call db.session.commit() after using this function to persist changes.
+
+ Args:
+ file_id (UUID): The ID of the associated file.
+ file_path (str): The path of the file.
+ file_content (dict): The content of the file.
+ operator_id (str): The ID of the operator performing the download.
+ operator_name (str): The name of the operator performing the download.
+
+ Returns:
+ DownloadHistory: The newly created download history record.
+
+ Raises:
+ DatabaseError:
+ If there is an error creating the download history record in the database.
"""
- file_record = Files()
- if file_id:
- file_record.id = file_id
- file_record.file_path = str(file_path)
- file_record.file_content = {
- "repositories": file_content.get("repositories", []),
- "groups": file_content.get("groups", []),
- "users": file_content.get("users", []),
- }
- db.session.add(file_record)
- db.session.commit()
- return file_record.id
+ try:
+ create_file(file_path, file_content, file_id)
+ download_history = DownloadHistory()
+ download_history.file_id = file_id
+ download_history.operator_id = operator_id
+ download_history.operator_name = operator_name
+ db.session.add(download_history)
+ except SQLAlchemyError as exc:
+ current_app.logger.error(str(exc))
+ error = E.FAILED_CREATE_DOWNLOAD_HISTORY_RECORD % {"file_id": file_id}
+ raise DatabaseError(error) from exc
+ return download_history
diff --git a/src/server/services/users.py b/src/server/services/users.py
index b1ac0b81..8b74bf09 100644
--- a/src/server/services/users.py
+++ b/src/server/services/users.py
@@ -7,7 +7,10 @@
import re
import typing as t
+from datetime import UTC, datetime
from http import HTTPStatus
+from pathlib import Path
+from uuid import uuid7
import requests
@@ -23,7 +26,9 @@
MAP_NO_RIGHTS_APPEND_PATTERN,
MAP_NO_RIGHTS_UPDATE_PATTERN,
MAP_NOT_FOUND_PATTERN,
+ USER_ROLES,
)
+from server.db import db
from server.entities.map_error import MapError
from server.entities.search_request import SearchResponse, SearchResult
from server.entities.summaries import UserSummary
@@ -32,6 +37,7 @@
ApiClientError,
ApiRequestError,
CredentialsError,
+ InvalidExportError,
InvalidFormError,
InvalidQueryError,
OAuthTokenError,
@@ -40,6 +46,9 @@
UnexpectedResponseError,
)
from server.messages import E, I
+from server.services import history_table
+from server.services.utils.affiliations import detect_affiliations
+from server.services.utils.permissions import get_permitted_repository_ids
from server.signals import user_deleted, user_updated
from .token import get_access_token, get_client_secret
@@ -55,6 +64,7 @@
if t.TYPE_CHECKING:
+ from server.api.schemas import FileQuery
from server.clients.users import UsersSearchResponse
from server.entities.map_user import Group, MapUser
from server.entities.patch_request import PatchOperation
@@ -733,3 +743,119 @@ def handle_user_updated(
users.get_by_id.clear_cache(user.id) # pyright: ignore[reportFunctionMemberAccess]
if user.eppns:
users.get_by_eppn.clear_cache(*user.eppns) # pyright: ignore[reportFunctionMemberAccess]
+
+
+def make_export_file(
+ user_ids: list[str], query: FileQuery, operator_id: str, operator_name: str
+) -> Path:
+ """Generate a file containing user details for the specified user IDs.
+
+ Args:
+ user_ids (list[str]): A list of user IDs to include in the export file.
+ query (FileQuery): The file query containing export format and other parameters.
+ operator_id (str): The ID of the operator performing the export.
+ operator_name (str): The name of the operator performing the export.
+
+ Returns:
+ Path: The path to the generated export file.
+ """
+ user_list = search(make_criteria_object("users", i=user_ids), raw=True).resources
+ delimiter = "," if query.f == "csv" else "\t"
+ file_id = uuid7()
+ target_dir = Path(config.STORAGE.local.storage) / datetime.now(UTC).strftime(
+ "%Y/%m"
+ )
+ target_dir.mkdir(parents=True, exist_ok=True)
+ file_path = target_dir / f"{file_id}.{query.f}"
+ file_path.write_text(delimiter.join(config.USERS.export_fields), encoding="utf-8")
+
+ permitted_repository_ids = get_permitted_repository_ids()
+
+ file_repositories, file_groups, file_users = _wite_user(
+ user_list, delimiter, file_path, permitted_repository_ids
+ )
+ file_content = {
+ "repositories": list(file_repositories),
+ "groups": list(file_groups),
+ "users": list(file_users),
+ }
+ history_table.create_download_history(
+ file_id, str(file_path), file_content, operator_id, operator_name
+ )
+ db.session.commit()
+ return file_path
+
+
+def _wite_user(
+ user_list: list[MapUser],
+ delimiter: str,
+ file_path: Path,
+ permitted_repository_ids: set[str],
+) -> tuple[set[dict[str, str]], set[dict[str, str]], set[dict[str, str]]]:
+ """Write user details to file.
+
+ Args:
+ user_list (list[MapUser]): A list of user details.
+ delimiter (str): The delimiter to use in the file.
+ file_path (Path): The path to the file.
+ permitted_repository_ids (list[str]): A list of permitted repository IDs.
+
+ Returns:
+ tuple[set[dict[str, str]], set[dict[str, str]], set[dict[str, str]]]:
+ A tuple containing sets of file repositories, file groups, and file users.
+
+ Raises:
+ InvalidExportError:
+ If the user cannot be exported due to insufficient permissions.
+ """
+ file_repositories = set[dict[str, str]]()
+ file_groups = set[dict[str, str]]()
+ file_users = set[dict[str, str]]()
+ for map_user in user_list:
+ roles, groups = detect_affiliations([g.value for g in map_user.groups or []])
+ if not is_current_user_system_admin() and any(
+ role.role == USER_ROLES.SYSTEM_ADMIN for role in roles
+ ):
+ error = E.USER_CANNOT_EXPORT_SYSTEM_ADMIN
+ raise InvalidExportError(error)
+ if not is_current_user_system_admin() and not any(
+ group.repository_id in permitted_repository_ids for group in groups
+ ):
+ error = E.USER_FORBIDDEN_EXPORT
+ raise InvalidExportError(error)
+
+ file_users.add({"id": map_user.id or "", "user_name": map_user.user_name or ""})
+ group_ids = []
+ for group in groups:
+ if group.repository_id not in permitted_repository_ids:
+ continue
+
+ file_groups.add({"id": group.group_id or "", "display_name": ""})
+ file_repositories.add({"id": group.repository_id or "", "display_name": ""})
+ group_ids.append(group.group_id or "")
+ roles_list = [
+ r.role.value for r in roles if r.repository_id in permitted_repository_ids
+ ] or [""]
+ eppns = [eppn.value for eppn in map_user.edu_person_principal_names or []]
+ emails = [email.value for email in map_user.emails or []]
+
+ max_len = max(len(group_ids), len(roles_list), len(eppns), len(emails))
+
+ for i in range(max_len):
+ row = [
+ map_user.id,
+ map_user.user_name,
+ group_ids[i] if i < len(group_ids) else group_ids[len(group_ids) - 1],
+ "", # group name is not exported. it can't be get from mAP Core API search user endpoint # noqa: E501
+ roles_list[i]
+ if i < len(roles_list)
+ else roles_list[len(roles_list) - 1],
+ eppns[i] if i < len(eppns) else eppns[len(eppns) - 1],
+ map_user.preferred_language or "",
+ emails[i] if i < len(emails) else emails[len(emails) - 1],
+ ]
+ file_path.write_text(
+ delimiter.join(row) + "\n",
+ encoding="utf-8",
+ )
+ return file_repositories, file_groups, file_users