Skip to content

Commit cbedeab

Browse files
author
maxim-lixakov
committed
[DOP-19788] - add filtering for Run
1 parent 2ad444b commit cbedeab

15 files changed

Lines changed: 354 additions & 132 deletions

File tree

‎docs/changelog/next_release/102.feature.rst‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
Add filters for **runs**

‎syncmaster/backend/api/v1/router.py‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
from syncmaster.backend.api.v1.connections import router as connection_router
77
from syncmaster.backend.api.v1.groups import router as group_router
88
from syncmaster.backend.api.v1.queue import router as queue_router
9+
from syncmaster.backend.api.v1.runs import router as runs_router
910
from syncmaster.backend.api.v1.transfers import router as transfer_router
1011
from syncmaster.backend.api.v1.users import router as user_router
1112

@@ -16,3 +17,4 @@
1617
router.include_router(connection_router)
1718
router.include_router(transfer_router)
1819
router.include_router(queue_router)
20+
router.include_router(runs_router)

‎syncmaster/backend/api/v1/runs.py‎

Lines changed: 150 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,150 @@
1+
# SPDX-FileCopyrightText: 2023-2024 MTS PJSC
2+
# SPDX-License-Identifier: Apache-2.0
3+
from datetime import datetime
4+
5+
from fastapi import APIRouter, Depends, Query
6+
from kombu.exceptions import KombuError
7+
8+
from syncmaster.backend.api.deps import UnitOfWorkMarker
9+
from syncmaster.backend.services import UnitOfWork, get_user
10+
from syncmaster.db.models import Status, User
11+
from syncmaster.db.utils import Permission
12+
from syncmaster.errors.registration import get_error_responses
13+
from syncmaster.exceptions.base import ActionNotAllowedError
14+
from syncmaster.exceptions.run import CannotConnectToTaskQueueError
15+
from syncmaster.exceptions.transfer import TransferNotFoundError
16+
from syncmaster.schemas.v1.connections.connection import ReadAuthDataSchema
17+
from syncmaster.schemas.v1.transfers.run import (
18+
CreateRunSchema,
19+
ReadRunSchema,
20+
RunPageSchema,
21+
)
22+
from syncmaster.worker.config import celery
23+
24+
router = APIRouter(tags=["Runs"], responses=get_error_responses())
25+
26+
27+
@router.get("/runs")
28+
async def read_runs(
29+
transfer_id: int,
30+
page: int = Query(gt=0, default=1),
31+
page_size: int = Query(gt=0, le=200, default=20),
32+
status: list[Status] | None = Query(default=None),
33+
started_at_since: datetime | None = Query(default=None),
34+
started_at_until: datetime | None = Query(default=None),
35+
current_user: User = Depends(get_user(is_active=True)),
36+
unit_of_work: UnitOfWork = Depends(UnitOfWorkMarker),
37+
) -> RunPageSchema:
38+
"""Return runs of transfer with pagination"""
39+
resource_rule = await unit_of_work.transfer.get_resource_permission(
40+
user=current_user,
41+
resource_id=transfer_id,
42+
)
43+
44+
if resource_rule == Permission.NONE:
45+
raise TransferNotFoundError
46+
47+
pagination = await unit_of_work.run.paginate(
48+
transfer_id=transfer_id,
49+
page=page,
50+
page_size=page_size,
51+
status=status,
52+
started_at_since=started_at_since,
53+
started_at_until=started_at_until,
54+
)
55+
56+
return RunPageSchema.from_pagination(pagination=pagination)
57+
58+
59+
@router.get("/runs/{run_id}")
60+
async def read_run(
61+
run_id: int,
62+
current_user: User = Depends(get_user(is_active=True)),
63+
unit_of_work: UnitOfWork = Depends(UnitOfWorkMarker),
64+
) -> ReadRunSchema:
65+
run = await unit_of_work.run.read_by_id(run_id=run_id)
66+
67+
resource_role = await unit_of_work.transfer.get_resource_permission(
68+
user=current_user,
69+
resource_id=run.transfer_id,
70+
)
71+
72+
if resource_role == Permission.NONE:
73+
raise TransferNotFoundError
74+
75+
return ReadRunSchema.from_orm(run)
76+
77+
78+
@router.post("/runs")
79+
async def start_run(
80+
create_run_data: CreateRunSchema,
81+
current_user: User = Depends(get_user(is_active=True)),
82+
unit_of_work: UnitOfWork = Depends(UnitOfWorkMarker),
83+
) -> ReadRunSchema:
84+
# Check: user can start transfer
85+
resource_rule = await unit_of_work.transfer.get_resource_permission(
86+
user=current_user,
87+
resource_id=create_run_data.transfer_id,
88+
)
89+
90+
if resource_rule == Permission.NONE:
91+
raise TransferNotFoundError
92+
93+
if resource_rule < Permission.WRITE:
94+
raise ActionNotAllowedError
95+
96+
transfer = await unit_of_work.transfer.read_by_id(transfer_id=create_run_data.transfer_id)
97+
98+
# The credentials.read method is used rather than credentials.read_bulk deliberately
99+
# it’s more convenient to transfer credits in this place
100+
credentials_source = await unit_of_work.credentials.read(
101+
transfer.source_connection_id,
102+
)
103+
credentials_target = await unit_of_work.credentials.read(
104+
transfer.target_connection_id,
105+
)
106+
107+
async with unit_of_work:
108+
run = await unit_of_work.run.create(
109+
transfer_id=create_run_data.transfer_id,
110+
# Since fields with credentials may have different names (for example, S3 and Postgres have different names)
111+
# the work of checking fields and removing passwords is delegated to the ReadAuthDataSchema class
112+
source_creds=ReadAuthDataSchema(auth_data=credentials_source).dict(),
113+
target_creds=ReadAuthDataSchema(auth_data=credentials_target).dict(),
114+
)
115+
try:
116+
celery.send_task("run_transfer_task", kwargs={"run_id": run.id}, queue=transfer.queue.name)
117+
except KombuError as e:
118+
async with unit_of_work:
119+
run = await unit_of_work.run.update(
120+
run_id=run.id,
121+
status=Status.FAILED,
122+
)
123+
raise CannotConnectToTaskQueueError(run_id=run.id) from e
124+
return ReadRunSchema.from_orm(run)
125+
126+
127+
@router.post("/runs/{run_id}/stop")
128+
async def stop_run(
129+
run_id: int,
130+
current_user: User = Depends(get_user(is_active=True)),
131+
unit_of_work: UnitOfWork = Depends(UnitOfWorkMarker),
132+
) -> ReadRunSchema:
133+
run = await unit_of_work.run.read_by_id(run_id=run_id)
134+
135+
# Check: user can stop transfer
136+
resource_rule = await unit_of_work.transfer.get_resource_permission(
137+
user=current_user,
138+
resource_id=run.transfer_id,
139+
)
140+
141+
if resource_rule == Permission.NONE:
142+
raise TransferNotFoundError
143+
144+
if resource_rule < Permission.WRITE:
145+
raise ActionNotAllowedError
146+
147+
async with unit_of_work:
148+
run = await unit_of_work.run.stop(run_id=run_id)
149+
# TODO: add immdiate stop transfer after stop Run
150+
return ReadRunSchema.from_orm(run)

‎syncmaster/backend/api/v1/transfers.py‎

Lines changed: 1 addition & 130 deletions
Original file line numberDiff line numberDiff line change
@@ -2,25 +2,22 @@
22
# SPDX-License-Identifier: Apache-2.0
33

44
from fastapi import APIRouter, Depends, Query, status
5-
from kombu.exceptions import KombuError
65

76
from syncmaster.backend.api.deps import UnitOfWorkMarker
87
from syncmaster.backend.services import UnitOfWork, get_user
9-
from syncmaster.db.models import Status, User
8+
from syncmaster.db.models import User
109
from syncmaster.db.utils import Permission
1110
from syncmaster.errors.registration import get_error_responses
1211
from syncmaster.exceptions.base import ActionNotAllowedError
1312
from syncmaster.exceptions.connection import ConnectionNotFoundError
1413
from syncmaster.exceptions.group import GroupNotFoundError
1514
from syncmaster.exceptions.queue import DifferentTransferAndQueueGroupError
16-
from syncmaster.exceptions.run import CannotConnectToTaskQueueError
1715
from syncmaster.exceptions.transfer import (
1816
DifferentTransferAndConnectionsGroupsError,
1917
DifferentTypeConnectionsAndParamsError,
2018
TransferNotFoundError,
2119
)
2220
from syncmaster.schemas.v1.connection_types import ConnectionType
23-
from syncmaster.schemas.v1.connections.connection import ReadAuthDataSchema
2421
from syncmaster.schemas.v1.status import (
2522
StatusCopyTransferResponseSchema,
2623
StatusResponseSchema,
@@ -32,12 +29,6 @@
3229
TransferPageSchema,
3330
UpdateTransferSchema,
3431
)
35-
from syncmaster.schemas.v1.transfers.run import (
36-
CreateRunSchema,
37-
ReadRunSchema,
38-
RunPageSchema,
39-
)
40-
from syncmaster.worker.config import celery
4132

4233
router = APIRouter(tags=["Transfers"], responses=get_error_responses())
4334

@@ -368,123 +359,3 @@ async def delete_transfer(
368359
status_code=status.HTTP_200_OK,
369360
message="Transfer was deleted",
370361
)
371-
372-
373-
@router.get("/runs")
374-
async def read_runs(
375-
transfer_id: int,
376-
page: int = Query(gt=0, default=1),
377-
page_size: int = Query(gt=0, le=200, default=20),
378-
current_user: User = Depends(get_user(is_active=True)),
379-
unit_of_work: UnitOfWork = Depends(UnitOfWorkMarker),
380-
) -> RunPageSchema:
381-
"""Return runs of transfer with pagination"""
382-
resource_rule = await unit_of_work.transfer.get_resource_permission(
383-
user=current_user,
384-
resource_id=transfer_id,
385-
)
386-
387-
if resource_rule == Permission.NONE:
388-
raise TransferNotFoundError
389-
390-
pagination = await unit_of_work.run.paginate(
391-
transfer_id=transfer_id,
392-
page=page,
393-
page_size=page_size,
394-
)
395-
396-
return RunPageSchema.from_pagination(pagination=pagination)
397-
398-
399-
@router.get("/runs/{run_id}")
400-
async def read_run(
401-
run_id: int,
402-
current_user: User = Depends(get_user(is_active=True)),
403-
unit_of_work: UnitOfWork = Depends(UnitOfWorkMarker),
404-
) -> ReadRunSchema:
405-
run = await unit_of_work.run.read_by_id(run_id=run_id)
406-
407-
resource_role = await unit_of_work.transfer.get_resource_permission(
408-
user=current_user,
409-
resource_id=run.transfer_id,
410-
)
411-
412-
if resource_role == Permission.NONE:
413-
raise TransferNotFoundError
414-
415-
return ReadRunSchema.from_orm(run)
416-
417-
418-
@router.post("/runs")
419-
async def start_run(
420-
create_run_data: CreateRunSchema,
421-
current_user: User = Depends(get_user(is_active=True)),
422-
unit_of_work: UnitOfWork = Depends(UnitOfWorkMarker),
423-
) -> ReadRunSchema:
424-
# Check: user can start transfer
425-
resource_rule = await unit_of_work.transfer.get_resource_permission(
426-
user=current_user,
427-
resource_id=create_run_data.transfer_id,
428-
)
429-
430-
if resource_rule == Permission.NONE:
431-
raise TransferNotFoundError
432-
433-
if resource_rule < Permission.WRITE:
434-
raise ActionNotAllowedError
435-
436-
transfer = await unit_of_work.transfer.read_by_id(transfer_id=create_run_data.transfer_id)
437-
438-
# The credentials.read method is used rather than credentials.read_bulk deliberately
439-
# it’s more convenient to transfer credits in this place
440-
credentials_source = await unit_of_work.credentials.read(
441-
transfer.source_connection_id,
442-
)
443-
credentials_target = await unit_of_work.credentials.read(
444-
transfer.target_connection_id,
445-
)
446-
447-
async with unit_of_work:
448-
run = await unit_of_work.run.create(
449-
transfer_id=create_run_data.transfer_id,
450-
# Since fields with credentials may have different names (for example, S3 and Postgres have different names)
451-
# the work of checking fields and removing passwords is delegated to the ReadAuthDataSchema class
452-
source_creds=ReadAuthDataSchema(auth_data=credentials_source).dict(),
453-
target_creds=ReadAuthDataSchema(auth_data=credentials_target).dict(),
454-
)
455-
try:
456-
celery.send_task("run_transfer_task", kwargs={"run_id": run.id}, queue=transfer.queue.name)
457-
except KombuError as e:
458-
async with unit_of_work:
459-
run = await unit_of_work.run.update(
460-
run_id=run.id,
461-
status=Status.FAILED,
462-
)
463-
raise CannotConnectToTaskQueueError(run_id=run.id) from e
464-
return ReadRunSchema.from_orm(run)
465-
466-
467-
@router.post("/runs/{run_id}/stop")
468-
async def stop_run(
469-
run_id: int,
470-
current_user: User = Depends(get_user(is_active=True)),
471-
unit_of_work: UnitOfWork = Depends(UnitOfWorkMarker),
472-
) -> ReadRunSchema:
473-
run = await unit_of_work.run.read_by_id(run_id=run_id)
474-
475-
# Check: user can stop transfer
476-
resource_rule = await unit_of_work.transfer.get_resource_permission(
477-
user=current_user,
478-
resource_id=run.transfer_id,
479-
)
480-
481-
if resource_rule == Permission.NONE:
482-
raise TransferNotFoundError
483-
484-
if resource_rule < Permission.WRITE:
485-
raise ActionNotAllowedError
486-
487-
async with unit_of_work:
488-
run = await unit_of_work.run.stop(run_id=run_id)
489-
# TODO: add immdiate stop transfer after stop Run
490-
return ReadRunSchema.from_orm(run)

‎syncmaster/db/repositories/run.py‎

Lines changed: 16 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
# SPDX-FileCopyrightText: 2023-2024 MTS PJSC
22
# SPDX-License-Identifier: Apache-2.0
3+
from datetime import datetime
34
from typing import Any, NoReturn
45

56
from sqlalchemy import desc, select
@@ -24,8 +25,22 @@ async def paginate(
2425
transfer_id: int,
2526
page: int,
2627
page_size: int,
28+
status: list[Status] | None = None,
29+
started_at_since: datetime | None = None,
30+
started_at_until: datetime | None = None,
2731
) -> Pagination:
28-
query = select(Run).where(Run.transfer_id == transfer_id).order_by(desc(Run.created_at))
32+
query = select(Run).where(Run.transfer_id == transfer_id)
33+
34+
if status:
35+
query = query.where(Run.status.in_(status))
36+
37+
if started_at_since:
38+
query = query.where(Run.started_at >= started_at_since)
39+
40+
if started_at_until:
41+
query = query.where(Run.started_at <= started_at_until)
42+
43+
query = query.order_by(desc(Run.created_at))
2944
return await self._paginate_scalar_result(query=query, page=page, page_size=page_size)
3045

3146
async def read_by_id(self, run_id: int) -> Run:

‎tests/conftest.py‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -46,7 +46,8 @@
4646
logger = logging.getLogger(__name__)
4747

4848
pytest_plugins = [
49-
"tests.test_unit.test_transfers.transfer_fixtures.group_transfers_fixture",
49+
"tests.test_unit.test_transfers.transfer_fixtures",
50+
"tests.test_unit.test_runs.run_fixtures",
5051
]
5152

5253

tests/test_unit/test_transfers/test_runs/__init__.py renamed to tests/test_unit/test_runs/__init__.py

File renamed without changes.

‎tests/test_unit/test_runs/run_fixtures/__init__.py‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,2 @@
1+
from tests.test_unit.test_runs.run_fixtures.group_run_fixture import group_run
2+
from tests.test_unit.test_runs.run_fixtures.group_runs_fixture import group_runs

tests/test_unit/test_transfers/test_runs/conftest.py renamed to tests/test_unit/test_runs/run_fixtures/group_run_fixture.py

File renamed without changes.

0 commit comments

Comments
 (0)