feat: settings, delete port validation

Merge pull request 'feat: endpoints services' (#21 ) from codestyle-issue-refactor into master
Reviewed-on: #21 Reviewed-by: Vladislav Syrochkin <vlad.dev@heado.ru> Reviewed-by: cyrussmeat <dr.cyrill@gmail.com>
2025-11-15 16:00:05 +05:00 · 2025-11-11 18:05:30 +05:00 · 2025-11-06 23:34:47 +05:00 · 2025-11-05 14:28:51 +05:00
24 changed files with 789 additions and 585 deletions
--- a/api/api/db/logic/keyring.py
+++ b/api/api/db/logic/keyring.py
@@ -37,7 +37,7 @@ async def update_key_by_id(connection: AsyncConnection, update_values, key) -> O
    await connection.commit()


-async def create_key(connection: AsyncConnection, key: AccountKeyring, key_id: int) -> Optional[AccountKeyring]:
+async def create_key(connection: AsyncConnection, key: AccountKeyring, key_id: str) -> Optional[AccountKeyring]:
    """
    Создает нове поле в таблице account_keyring_table).
    """
--- a/api/api/db/logic/list_events.py
+++ b/api/api/db/logic/list_events.py
@@ -121,100 +121,6 @@ async def get_list_events_page_DTO(
    )


-async def get_list_events_page_by_creator_id(
-    connection: AsyncConnection, creator_id: int, page: int, limit: int
-) -> Optional[AllListEventResponse]:
-    """
-    Получает список событий заданного создателя по значениям page и limit и creator_id.
-    """
-
-    first_event = page * limit - limit
-    query = (
-        select(
-            list_events_table.c.id,
-            list_events_table.c.name,
-            list_events_table.c.title,
-            list_events_table.c.creator_id,
-            list_events_table.c.created_at,
-            list_events_table.c.schema,
-            list_events_table.c.state,
-            list_events_table.c.status,
-        )
-        .where(list_events_table.c.creator_id == creator_id)  # Фильтрация по creator_id
-        .order_by(list_events_table.c.id)
-        .offset(first_event)
-        .limit(limit)
-    )
-
-    count_query = (
-        select(func.count())
-        .select_from(list_events_table)
-        .where(list_events_table.c.creator_id == creator_id)  # Фильтрация по creator_id
-    )
-
-    result = await connection.execute(query)
-    count_result = await connection.execute(count_query)
-
-    events_data = result.mappings().all()
-    total_count = count_result.scalar()
-    total_pages = math.ceil(total_count / limit)
-
-    # Здесь предполагается, что all_list_event_adapter.validate_python корректно обрабатывает данные
-    validated_list_event = all_list_event_adapter.validate_python(events_data)
-
-    return AllListEventResponse(
-        list_event=validated_list_event,
-        amount_count=total_count,
-        amount_pages=total_pages,
-        current_page=page,
-        limit=limit,
-    )
-
-
-async def get_list_events_page(connection: AsyncConnection, page, limit) -> Optional[AllListEventResponse]:
-    """
-    Получает список событий заданного создателя по значениям page и limit.
-    """
-
-    first_event = page * limit - (limit)
-
-    query = (
-        select(
-            list_events_table.c.id,
-            list_events_table.c.name,
-            list_events_table.c.title,
-            list_events_table.c.creator_id,
-            list_events_table.c.created_at,
-            list_events_table.c.schema,
-            list_events_table.c.state,
-            list_events_table.c.status,
-        )
-        .order_by(list_events_table.c.id)
-        .offset(first_event)
-        .limit(limit)
-    )
-
-    count_query = select(func.count()).select_from(list_events_table)
-
-    result = await connection.execute(query)
-    count_result = await connection.execute(count_query)
-
-    events_data = result.mappings().all()
-    total_count = count_result.scalar()
-    total_pages = math.ceil(total_count / limit)
-
-    # Здесь предполагается, что all_list_event_adapter.validate_python корректно обрабатывает данные
-    validated_list_event = all_list_event_adapter.validate_python(events_data)
-
-    return AllListEventResponse(
-        list_event=validated_list_event,
-        amount_count=total_count,
-        amount_pages=total_pages,
-        current_page=page,
-        limit=limit,
-    )
-
-
 async def get_list_events_by_name(connection: AsyncConnection, name: str) -> Optional[ListEvent]:
    """
    Получает list events  по name.
--- a/api/api/db/logic/node_link.py
+++ b/api/api/db/logic/node_link.py
@@ -1,4 +1,4 @@
-from typing import Optional
+from typing import Optional, List

 from datetime import datetime, timezone

@@ -79,3 +79,20 @@ async def create_node_link_schema(
    await connection.commit()

    return await get_last_node_link_by_creator_and_ps_id(connection, creator_id, validated_link_schema.from_id)
+
+
+async def get_all_node_links_by_next_node_ids(connection: AsyncConnection, next_node_ids: List[int]) -> List[NodeLink]:
+    """
+    Получает все активные node_link для списка next_node_id одним запросом.
+    """
+    if not next_node_ids:
+        return []
+
+    query = select(node_link_table).where(
+        node_link_table.c.next_node_id.in_(next_node_ids), node_link_table.c.status == NodeLinkStatus.ACTIVE.value
+    )
+
+    node_link_db_cursor = await connection.execute(query)
+    node_links_data = node_link_db_cursor.mappings().all()
+
+    return [NodeLink.model_validate(link_data) for link_data in node_links_data]
--- a/api/api/db/logic/process_schema.py
+++ b/api/api/db/logic/process_schema.py
@@ -141,51 +141,6 @@ async def update_process_schema_by_id(connection: AsyncConnection, update_values
    await connection.commit()


-async def update_process_schema_settings_by_id(
-    connection: AsyncConnection, process_schema_id: int, node_data: Dict[str, Any]
-):
-    """
-    Добавляет новый узел в массив 'nodes' в настройках процесса.
-    Если массив 'nodes' не существует, создает его.
-    """
-    # Получаем текущие settings
-    query = select(process_schema_table.c.settings).where(process_schema_table.c.id == process_schema_id)
-    result = await connection.execute(query)
-    current_settings = result.scalar_one_or_none()
-
-    # Если settings пустые, создаем пустой словарь
-    if current_settings is None:
-        current_settings = {}
-
-    # Инициализируем массив nodes, если его нет
-    if "nodes" not in current_settings:
-        current_settings["nodes"] = []
-
-    # Добавляем новый узел в массив
-    current_settings["nodes"].append(node_data)
-
-    # Обновляем поле settings
-    await connection.execute(
-        process_schema_table.update()
-        .where(process_schema_table.c.id == process_schema_id)
-        .values(settings=current_settings)
-    )
-
-    await connection.commit()
-
-
-async def get_last_created_process_schema(connection: AsyncConnection) -> Optional[int]:
-    """
-    Получает ID последней созданной схемы процесса.
-    """
-    query = select(process_schema_table.c.id).order_by(desc(process_schema_table.c.id)).limit(1)
-
-    result = await connection.execute(query)
-    last_id = result.scalar_one_or_none()
-
-    return last_id
-
-
 async def create_process_schema(
    connection: AsyncConnection, creator_id: int, title: str, description: str
 ) -> Optional[int]:
--- a/api/api/db/logic/ps_node.py
+++ b/api/api/db/logic/ps_node.py
@@ -1,4 +1,4 @@
-from typing import Optional, List
+from typing import Optional, List, Dict, Any

 from datetime import datetime, timezone

@@ -8,7 +8,6 @@ from sqlalchemy.ext.asyncio import AsyncConnection
 from orm.tables.process import ps_node_table, node_link_table, process_schema_table

 from api.schemas.process.ps_node import Ps_Node
-from model_nodes.node_listen_models import ListenNodeCoreSchema
 from orm.tables.process import NodeStatus


@@ -49,18 +48,34 @@ async def get_last_ps_node_by_creator_and_ps_id(
    return Ps_Node.model_validate(ps_node_data)


+async def get_all_ps_nodes_by_ps_id(connection: AsyncConnection, ps_id: int) -> List[Ps_Node]:
+    """
+    Получает все активные ps_node для данной process_schema.
+    """
+    query = select(ps_node_table).where(
+        ps_node_table.c.ps_id == ps_id, ps_node_table.c.status == NodeStatus.ACTIVE.value
+    )
+
+    ps_node_db_cursor = await connection.execute(query)
+    ps_nodes_data = ps_node_db_cursor.mappings().all()
+
+    return [Ps_Node.model_validate(node_data) for node_data in ps_nodes_data]
+
+
 async def create_ps_node_schema(
    connection: AsyncConnection,
    validated_schema,
+    node_descriptor,
    creator_id: int,
-) -> Optional[ListenNodeCoreSchema]:
+    settings_payload: Optional[Dict[str, Any]] = None,
+) -> Optional[Ps_Node]:
    """
    Создает нове поле в таблице process_schema_table.
    """
    query = insert(ps_node_table).values(
        ps_id=validated_schema.ps_id,
        node_type=validated_schema.node_type,
-        settings=validated_schema.data.model_dump(),
+        settings=settings_payload if settings_payload is not None else node_descriptor.model_dump(),
        creator_id=creator_id,
        created_at=datetime.now(timezone.utc),
        status=NodeStatus.ACTIVE.value,
--- a/api/api/endpoints/account.py
+++ b/api/api/endpoints/account.py
@@ -1,23 +1,17 @@
 from typing import List, Optional

-from fastapi import APIRouter, Depends, HTTPException, Query, status
-from orm.tables.account import AccountStatus
+from fastapi import APIRouter, Depends, Query, status
 from sqlalchemy.ext.asyncio import AsyncConnection

 from api.db.connection.session import get_connection_dep
-from api.db.logic.account import (
-    create_user,
-    get_user_account_page_DTO,
-    get_user_by_id,
-    get_user_by_login,
-    update_user_by_id,
-)
-from api.db.logic.keyring import create_password_key, update_password_key
+from api.db.logic.account import get_user_by_login
 from api.schemas.account.account import User
 from api.schemas.base import bearer_schema
-from api.schemas.endpoints.account import AllUserResponse, UserCreate, UserFilterDTO, UserUpdate
+from api.schemas.endpoints.account import AllUserResponse, UserCreate, UserUpdate, UserFilterDTO
 from api.services.auth import get_current_user
 from api.services.user_role_validation import UserRoleValidator
+from api.error import create_operation_error, create_validation_error
+from api.services.endpoints.account import AccountService

 api_router = APIRouter(
    prefix="/account",
@@ -39,7 +33,7 @@ async def get_all_account_endpoint(
    current_user=Depends(get_current_user),
 ):
    validator = UserRoleValidator(connection)
-    authorize_user = await validator.validate_admin(current_user)
+    await validator.validate_admin(current_user)

    filters = {
        **({"status": status_filter} if status_filter else {}),
@@ -54,10 +48,14 @@ async def get_all_account_endpoint(
        filters=filters if filters else None,
    )

-    user_list = await get_user_account_page_DTO(connection, filter_dto)
+    service = AccountService(connection)
+    user_list = await service.list(filter_dto)

    if user_list is None:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Accounts not found")
+        raise create_operation_error(
+            message="Accounts not found",
+            status_code=status.HTTP_404_NOT_FOUND,
+        )

    return user_list

@@ -69,12 +67,17 @@ async def get_account_endpoint(
    current_user=Depends(get_current_user),
 ):
    validator = UserRoleValidator(connection)
-    authorize_user = await validator.validate_admin(current_user)
+    await validator.validate_admin(current_user)

-    user = await get_user_by_id(connection, user_id)
+    service = AccountService(connection)
+    user = await service.get(user_id)

    if user is None:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Account not found")
+        raise create_operation_error(
+            message="Account not found",
+            status_code=status.HTTP_404_NOT_FOUND,
+            details={"user_id": user_id},
+        )

    return user

@@ -91,12 +94,14 @@ async def create_account_endpoint(
    user_validation = await get_user_by_login(connection, user.login)

    if user_validation is None:
-        new_user = await create_user(connection, user, authorize_user.id)
-        await create_password_key(connection, user.password, new_user.id)
+        service = AccountService(connection)
+        new_user = await service.create(user_data=user, creator_id=authorize_user.id)
        return new_user
    else:
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST, detail="An account with this information already exists."
+        raise create_validation_error(
+            message="An account with this information already exists.",
+            status_code=status.HTTP_400_BAD_REQUEST,
+            details={"login": user.login},
        )


@@ -108,25 +113,30 @@ async def update_account_endpoint(
    current_user=Depends(get_current_user),
 ):
    validator = UserRoleValidator(connection)
-    authorize_user = await validator.validate_admin(current_user)
+    await validator.validate_admin(current_user)
+
+    service = AccountService(connection)
+    user = await service.get(user_id)

-    user = await get_user_by_id(connection, user_id)
    if user is None:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Account not found")
-
-    if user_update.password is not None:
-        await update_password_key(connection, user.id, user_update.password)
+        raise create_operation_error(
+            message="Account not found",
+            status_code=status.HTTP_404_NOT_FOUND,
+            details={"user_id": user_id},
+        )

    updated_values = user_update.model_dump(by_alias=True, exclude_none=True)

    if not updated_values:
        return user

-    await update_user_by_id(connection, updated_values, user)
+    updated_user = await service.update(
+        user_id=user_id,
+        user_update_data=updated_values,
+        password=user_update.password,
+    )

-    user = await get_user_by_id(connection, user_id)
-
-    return user
+    return updated_user


@api_router.delete("/{user_id}", dependencies=[Depends(bearer_schema)], response_model=User)
@@ -136,21 +146,17 @@ async def delete_account_endpoint(
    current_user=Depends(get_current_user),
 ):
    validator = UserRoleValidator(connection)
-    authorize_user = await validator.validate_admin(current_user)
+    await validator.validate_admin(current_user)
+
+    service = AccountService(connection)
+    user = await service.get(user_id)

-    user = await get_user_by_id(connection, user_id)
    if user is None:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Account not found")
+        raise create_operation_error(
+            message="Account not found",
+            status_code=status.HTTP_404_NOT_FOUND,
+            details={"user_id": user_id},
+        )
+    deleted_user = await service.delete(user_id=user_id)

-    user_update = UserUpdate(status=AccountStatus.DELETED.value)
-
-    updated_values = user_update.model_dump(by_alias=True, exclude_none=True)
-
-    if not updated_values:
-        return user
-
-    await update_user_by_id(connection, updated_values, user)
-
-    user = await get_user_by_id(connection, user_id)
-
-    return user
+    return deleted_user
--- a/api/api/endpoints/auth.py
+++ b/api/api/endpoints/auth.py
@@ -1,15 +1,11 @@
-from datetime import datetime, timedelta, timezone
-
 from fastapi import (
    APIRouter,
    Depends,
-    HTTPException,
    Response,
    status,
    Request,
 )

-from loguru import logger
 from fastapi_jwt_auth import AuthJWT

 from pydantic import BaseModel
@@ -19,10 +15,10 @@ from sqlalchemy.ext.asyncio import AsyncConnection
 from api.config import get_settings
 from api.db.connection.session import get_connection_dep
 from api.services.auth import authenticate_user
-
-from api.db.logic.auth import add_new_refresh_token, upgrade_old_refresh_token
+from api.services.endpoints.auth import AuthService

 from api.schemas.endpoints.auth import Auth, Tokens
+from api.error import create_access_error

 api_router = APIRouter(
    prefix="/auth",
@@ -57,30 +53,18 @@ async def login_for_access_token_endpoint(
 ):
    """Авторизирует, выставляет токены в куки."""

-    user = await authenticate_user(connection, user.login, user.password)
+    authenticated_user = await authenticate_user(connection, user.login, user.password)

-    # print("login_for_access_token", user)
-
-    if not user:
-        raise HTTPException(
+    if not authenticated_user:
+        raise create_access_error(
+            message="Incorrect username or password",
            status_code=status.HTTP_401_UNAUTHORIZED,
-            detail="Incorrect username or password",
-            # headers={"WWW-Authenticate": "Bearer"},
        )

-    access_token_expires = timedelta(minutes=get_settings().ACCESS_TOKEN_EXPIRE_MINUTES)
-    refresh_token_expires = timedelta(days=get_settings().REFRESH_TOKEN_EXPIRE_DAYS)
+    service = AuthService(connection)
+    tokens = await service.login(authenticated_user, Authorize)

-    logger.debug(f"refresh_token_expires {refresh_token_expires}")
-
-    access_token = Authorize.create_access_token(subject=user.login, expires_time=access_token_expires)
-    refresh_token = Authorize.create_refresh_token(subject=user.login, expires_time=refresh_token_expires)
-
-    refresh_token_expires_time = datetime.now(timezone.utc) + refresh_token_expires
-
-    await add_new_refresh_token(connection, refresh_token, refresh_token_expires_time, user)
-
-    return Tokens(access_token=access_token, refresh_token=refresh_token)
+    return tokens


@api_router.post("/refresh", response_model=Tokens)
@@ -89,18 +73,19 @@ async def refresh_endpoint(
    connection: AsyncConnection = Depends(get_connection_dep),
    Authorize: AuthJWT = Depends(),
 ) -> Tokens:
+    service = AuthService(connection)
+
    try:
        Authorize.jwt_refresh_token_required()
        current_user = Authorize.get_jwt_subject()
    except Exception:
        refresh_token = request.headers.get("Authorization").split(" ")[1]
-        await upgrade_old_refresh_token(connection, refresh_token)
-        raise HTTPException(
+        await service.invalidate_refresh_token(refresh_token)
+        raise create_access_error(
+            message="Invalid refresh token",
            status_code=status.HTTP_401_UNAUTHORIZED,
-            detail="Invalid refresh token",
        )

-    access_token_expires = timedelta(minutes=get_settings().ACCESS_TOKEN_EXPIRE_MINUTES)
-    new_access_token = Authorize.create_access_token(subject=current_user, expires_time=access_token_expires)
+    tokens = await service.refresh(current_user, Authorize)

-    return Tokens(access_token=new_access_token)
+    return tokens
--- a/api/api/endpoints/keyring.py
+++ b/api/api/endpoints/keyring.py
@@ -1,19 +1,18 @@
 from fastapi import (
    APIRouter,
    Depends,
-    HTTPException,
    status,
 )
-from orm.tables.account import KeyStatus
 from sqlalchemy.ext.asyncio import AsyncConnection

 from api.db.connection.session import get_connection_dep
-from api.db.logic.keyring import create_key, get_key_by_id, update_key_by_id
 from api.schemas.account.account_keyring import AccountKeyring
 from api.schemas.base import bearer_schema
 from api.schemas.endpoints.account_keyring import AccountKeyringUpdate
 from api.services.auth import get_current_user
 from api.services.user_role_validation import UserRoleValidator
+from api.error import create_operation_error, create_validation_error
+from api.services.endpoints.keyring import KeyringService

 api_router = APIRouter(
    prefix="/keyring",
@@ -26,12 +25,17 @@ async def get_keyring_endpoint(
    key_id: str, connection: AsyncConnection = Depends(get_connection_dep), current_user=Depends(get_current_user)
 ):
    validator = UserRoleValidator(connection)
-    authorize_user = await validator.validate_admin(current_user)
+    await validator.validate_admin(current_user)

-    keyring = await get_key_by_id(connection, key_id)
+    service = KeyringService(connection)
+    keyring = await service.get(key_id)

    if keyring is None:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Key not found")
+        raise create_operation_error(
+            message="Key not found",
+            status_code=status.HTTP_404_NOT_FOUND,
+            details={"key_id": key_id},
+        )

    return keyring

@@ -45,21 +49,20 @@ async def create_keyring_endpoint(
    current_user=Depends(get_current_user),
 ):
    validator = UserRoleValidator(connection)
-    authorize_user = await validator.validate_admin(current_user)
+    await validator.validate_admin(current_user)

-    keyring = await get_key_by_id(connection, key_id)
+    service = KeyringService(connection)
+    keyring = await service.get(key_id)

    if keyring is None:
-        keyring_new = await create_key(
-            connection,
-            key,
-            key_id,
-        )
+        keyring_new = await service.create(key, key_id)
        return keyring_new

    else:
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST, detail="An keyring with this information already exists."
+        raise create_validation_error(
+            message="A keyring with this information already exists.",
+            status_code=status.HTTP_400_BAD_REQUEST,
+            details={"key_id": key_id},
        )


@@ -72,22 +75,26 @@ async def update_keyring_endpoint(
    current_user=Depends(get_current_user),
 ):
    validator = UserRoleValidator(connection)
-    authorize_user = await validator.validate_admin(current_user)
+    await validator.validate_admin(current_user)
+
+    service = KeyringService(connection)
+    keyring = await service.get(key_id)

-    keyring = await get_key_by_id(connection, key_id)
    if keyring is None:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="keyring not found")
+        raise create_operation_error(
+            message="keyring not found",
+            status_code=status.HTTP_404_NOT_FOUND,
+            details={"key_id": key_id},
+        )

    updated_values = keyring_update.model_dump(by_alias=True, exclude_none=True)

    if not updated_values:
        return keyring

-    await update_key_by_id(connection, updated_values, keyring)
+    updated_keyring = await service.update(key_id, updated_values)

-    keyring = await get_key_by_id(connection, key_id)
-
-    return keyring
+    return updated_keyring


@api_router.delete("/{user_id}/{key_id}", dependencies=[Depends(bearer_schema)], response_model=AccountKeyring)
@@ -98,21 +105,18 @@ async def delete_keyring_endpoint(
    current_user=Depends(get_current_user),
 ):
    validator = UserRoleValidator(connection)
-    authorize_user = await validator.validate_admin(current_user)
+    await validator.validate_admin(current_user)
+
+    service = KeyringService(connection)
+    keyring = await service.get(key_id)

-    keyring = await get_key_by_id(connection, key_id)
    if keyring is None:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="keyring not found")
+        raise create_operation_error(
+            message="keyring not found",
+            status_code=status.HTTP_404_NOT_FOUND,
+            details={"key_id": key_id},
+        )

-    keyring_update = AccountKeyringUpdate(status=KeyStatus.DELETED.value)
+    deleted_keyring = await service.delete(key_id)

-    updated_values = keyring_update.model_dump(by_alias=True, exclude_none=True)
-
-    if not updated_values:
-        return keyring
-
-    await update_key_by_id(connection, updated_values, keyring)
-
-    keyring = await get_key_by_id(connection, key_id)
-
-    return keyring
+    return deleted_keyring
--- a/api/api/endpoints/list_events.py
+++ b/api/api/endpoints/list_events.py
@@ -1,23 +1,17 @@
 from typing import List, Optional

-from fastapi import APIRouter, Depends, HTTPException, Query, status
-from orm.tables.events import EventStatus
+from fastapi import APIRouter, Depends, Query, status
 from sqlalchemy.ext.asyncio import AsyncConnection

 from api.db.connection.session import get_connection_dep
 from api.db.logic.account import get_user_by_login
-from api.db.logic.list_events import (
-    create_list_events,
-    get_list_events_by_id,
-    get_list_events_by_name,
-    get_list_events_page_DTO,
-    update_list_events_by_id,
-)
 from api.schemas.base import bearer_schema
 from api.schemas.endpoints.list_events import AllListEventResponse, ListEventFilterDTO, ListEventUpdate
 from api.schemas.events.list_events import ListEvent
 from api.services.auth import get_current_user
 from api.services.user_role_validation import UserRoleValidator
+from api.error import create_operation_error, create_validation_error
+from api.services.endpoints.list_events import ListEventsService

 api_router = APIRouter(
    prefix="/list_events",
@@ -59,10 +53,14 @@ async def get_all_list_events_endpoint(
            filter_dto.filters = {}
        filter_dto.filters["creator_id"] = [str(authorize_user.id)]

-    list_events_page = await get_list_events_page_DTO(connection, filter_dto)
+    service = ListEventsService(connection)
+    list_events_page = await service.list(filter_dto)

    if list_events_page is None:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="List events not found")
+        raise create_operation_error(
+            message="List events not found",
+            status_code=status.HTTP_404_NOT_FOUND,
+        )

    return list_events_page

@@ -73,16 +71,18 @@ async def get_list_events_endpoint(
    connection: AsyncConnection = Depends(get_connection_dep),
    current_user=Depends(get_current_user),
 ):
-    list_events_validation = await get_list_events_by_id(connection, list_events_id)
+    service = ListEventsService(connection)
+    list_events_validation = await service.get(list_events_id)

    if list_events_validation is None:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="List events not found")
+        raise create_operation_error(
+            message="List events not found",
+            status_code=status.HTTP_404_NOT_FOUND,
+            details={"list_events_id": list_events_id},
+        )

    validator = UserRoleValidator(connection)
-    authorize_user = await validator.validate_ownership(current_user, list_events_validation.creator_id)
-
-    if list_events_id is None:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="List events not found")
+    await validator.validate_ownership(current_user, list_events_validation.creator_id)

    return list_events_validation

@@ -93,19 +93,34 @@ async def create_list_events_endpoint(
    connection: AsyncConnection = Depends(get_connection_dep),
    current_user=Depends(get_current_user),
 ):
-    user_validation = await get_user_by_login(connection, current_user)
-    list_events_validation = await get_list_events_by_name(connection, list_events.name)
-
-    if list_events_validation is None:
-        await create_list_events(connection, list_events, user_validation.id)
-        list_events_new = await get_list_events_by_name(connection, list_events.name)
-        return list_events_new
-
-    else:
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST, detail="An List events with this information already exists."
+    if list_events.name is None:
+        raise create_validation_error(
+            message="Name is required for list event creation",
+            status_code=status.HTTP_400_BAD_REQUEST,
        )

+    user_validation = await get_user_by_login(connection, current_user)
+
+    service = ListEventsService(connection)
+    list_events_validation = await service.get_by_name(list_events.name)
+
+    if list_events_validation is not None:
+        raise create_validation_error(
+            message="A List events with this information already exists.",
+            status_code=status.HTTP_400_BAD_REQUEST,
+            details={"name": list_events.name},
+        )
+
+    list_events_new = await service.create(list_events, user_validation.id)
+
+    if list_events_new is None:
+        raise create_operation_error(
+            message="Failed to create list event",
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+        )
+
+    return list_events_new
+

@api_router.put("/{list_events_id}", dependencies=[Depends(bearer_schema)], response_model=ListEvent)
 async def update_list_events(
@@ -114,24 +129,27 @@ async def update_list_events(
    connection: AsyncConnection = Depends(get_connection_dep),
    current_user=Depends(get_current_user),
 ):
-    list_events_validation = await get_list_events_by_id(connection, list_events_id)
+    service = ListEventsService(connection)
+    list_events_validation = await service.get(list_events_id)

    if list_events_validation is None:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="List events not found")
+        raise create_operation_error(
+            message="List events not found",
+            status_code=status.HTTP_404_NOT_FOUND,
+            details={"list_events_id": list_events_id},
+        )

    validator = UserRoleValidator(connection)
-    authorize_user = await validator.validate_ownership(current_user, list_events_validation.creator_id)
+    await validator.validate_ownership(current_user, list_events_validation.creator_id)

    updated_values = list_events_update.model_dump(by_alias=True, exclude_none=True)

    if not updated_values:
        return list_events_validation

-    await update_list_events_by_id(connection, updated_values, list_events_validation)
+    updated_list_events = await service.update(list_events_id, updated_values)

-    list_events = await get_list_events_by_id(connection, list_events_id)
-
-    return list_events
+    return updated_list_events


@api_router.delete("/{list_events_id}", dependencies=[Depends(bearer_schema)], response_model=ListEvent)
@@ -140,23 +158,19 @@ async def delete_list_events_endpoint(
    connection: AsyncConnection = Depends(get_connection_dep),
    current_user=Depends(get_current_user),
 ):
-    list_events_validation = await get_list_events_by_id(connection, list_events_id)
+    service = ListEventsService(connection)
+    list_events_validation = await service.get(list_events_id)

    if list_events_validation is None:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="List events not found")
+        raise create_operation_error(
+            message="List events not found",
+            status_code=status.HTTP_404_NOT_FOUND,
+            details={"list_events_id": list_events_id},
+        )

    validator = UserRoleValidator(connection)
-    authorize_user = await validator.validate_ownership(current_user, list_events_validation.creator_id)
+    await validator.validate_ownership(current_user, list_events_validation.creator_id)

-    list_events_update = ListEventUpdate(status=EventStatus.DELETED.value)
+    deleted_list_events = await service.delete(list_events_id)

-    updated_values = list_events_update.model_dump(by_alias=True, exclude_none=True)
-
-    if not updated_values:
-        return list_events_validation
-
-    await update_list_events_by_id(connection, updated_values, list_events_validation)
-
-    list_events = await get_list_events_by_id(connection, list_events_id)
-
-    return list_events
+    return deleted_list_events
--- a/api/api/endpoints/process_schema.py
+++ b/api/api/endpoints/process_schema.py
@@ -1,38 +1,18 @@
 from typing import List, Optional

-from fastapi import APIRouter, Depends, HTTPException, Query, status
-from orm.tables.process import ProcessStatus
+from fastapi import APIRouter, Depends, Query, status, HTTPException
 from sqlalchemy.ext.asyncio import AsyncConnection

 from api.db.connection.session import get_connection_dep
 from api.db.logic.account import get_user_by_login
-from api.db.logic.process_schema import (
-    create_process_schema,
-    get_process_schema_by_id,
-    get_process_schema_page_DTO,
-    update_process_schema_by_id,
-)
 from api.schemas.base import bearer_schema
 from api.schemas.endpoints.process_schema import AllProcessSchemaResponse, ProcessSchemaFilterDTO, ProcessSchemaUpdate
-from api.schemas.process.process_schema import ProcessSchema, ProcessSchemaSettingsNode, ProcessSchemaResponse
-from api.schemas.process.ps_node import Ps_NodeFrontResponseNode
-from api.schemas.process.ps_node import Ps_NodeFrontResponse
+from api.schemas.process.process_schema import ProcessSchema, ProcessSchemaResponse
 from api.services.auth import get_current_user
 from api.services.user_role_validation import UserRoleValidator
-
-
-from api.db.logic.ps_node import create_ps_node_schema
-from api.db.logic.process_schema import update_process_schema_settings_by_id
-
-from orm.tables.process import NodeType
-
 from api.utils.to_camel_dict import to_camel_dict
-
-from core import VorkNodeRegistry
-
-from model_nodes import ListenNodeData
-
-from api.utils.node_counter import increment_node_counter
+from api.error import create_operation_error
+from api.services.endpoints.process_schema import ProcessSchemaService


 api_router = APIRouter(
@@ -83,32 +63,38 @@ async def get_all_process_schema_endpoint(
            filter_dto.filters = {}
        filter_dto.filters["creator_id"] = [str(authorize_user.id)]

-    process_schema_page = await get_process_schema_page_DTO(connection, filter_dto)
+    service = ProcessSchemaService(connection)
+    process_schema_page = await service.list(filter_dto)

    if process_schema_page is None:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Process schema not found")
+        raise create_operation_error(
+            message="Process schema not found",
+            status_code=status.HTTP_404_NOT_FOUND,
+        )

    return to_camel_dict(process_schema_page.model_dump())


-@api_router.get("/{process_schema_id}", dependencies=[Depends(bearer_schema)], response_model=ProcessSchema)
+@api_router.get("/{process_schema_id}", dependencies=[Depends(bearer_schema)], response_model=ProcessSchemaResponse)
 async def get_process_schema_endpoint(
    process_schema_id: int,
    connection: AsyncConnection = Depends(get_connection_dep),
    current_user=Depends(get_current_user),
 ):
-    process_schema_validation = await get_process_schema_by_id(connection, process_schema_id)
+    service = ProcessSchemaService(connection)
+    process_schema_response = await service.get(process_schema_id)

-    if process_schema_validation is None:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Process schema not found")
+    if process_schema_response is None:
+        raise create_operation_error(
+            message="Process schema not found",
+            status_code=status.HTTP_404_NOT_FOUND,
+            details={"process_schema_id": process_schema_id},
+        )

    validator = UserRoleValidator(connection)
-    authorize_user = await validator.validate_ownership(current_user, process_schema_validation.creator_id)
+    await validator.validate_ownership(current_user, process_schema_response.process_schema.creator_id)

-    if process_schema_id is None:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Process schema not found")
-
-    return to_camel_dict(process_schema_validation.model_dump())
+    return to_camel_dict(process_schema_response.model_dump())


@api_router.post("", dependencies=[Depends(bearer_schema)], response_model=ProcessSchemaResponse)
@@ -118,57 +104,8 @@ async def create_processschema_endpoint(
 ):
    user_validation = await get_user_by_login(connection, current_user)

-    current_node_counter = increment_node_counter()
-    title = f"Новая схема {current_node_counter}"
-
-    description = "Default description"
-
-    node_id = await create_process_schema(connection, user_validation.id, title, description)
-
-    process_schema_new = await get_process_schema_by_id(connection, node_id)
-
-    start_node_data = ListenNodeData(ps_id=process_schema_new.id, node_type=NodeType.START.value, is_start="True")
-
-    start_node_links = {}
-
-    registery = VorkNodeRegistry()
-
-    vork_node = registery.get("LISTEN")
-
-    node_descriptor = vork_node.form()
-
-    start_node = vork_node(data=start_node_data.model_dump(), links=start_node_links)
-
-    validated_start_schema = start_node.validate()
-
-    db_start_schema = await create_ps_node_schema(connection, validated_start_schema, user_validation.id)
-
-    node = ProcessSchemaSettingsNode(
-        id=db_start_schema.id,
-        node_type=NodeType.LISTEN.value,
-        data=validated_start_schema.data.model_dump(),
-        from_node=None,
-        links=None,
-    )
-
-    settings_dict = {"node": node.model_dump(mode="json")}
-
-    await update_process_schema_settings_by_id(connection, process_schema_new.id, settings_dict)
-
-    process_schema_new = await get_process_schema_by_id(connection, node_id)
-
-    ps_node_front_response = Ps_NodeFrontResponse(
-        description=node_descriptor.model_dump(),
-        node=Ps_NodeFrontResponseNode(
-            id=db_start_schema.id, node_type=NodeType.LISTEN.value, data=validated_start_schema.data.model_dump()
-        ),
-        link=None,
-    )
-
-    response_data = {
-        "process_schema": process_schema_new.model_dump(),
-        "node_listen": ps_node_front_response.model_dump(),
-    }
+    service = ProcessSchemaService(connection)
+    response_data = await service.create(user_validation.id)

    return to_camel_dict(response_data)

@@ -180,22 +117,25 @@ async def update_process_schema_endpoint(
    connection: AsyncConnection = Depends(get_connection_dep),
    current_user=Depends(get_current_user),
 ):
-    process_schema_validation = await get_process_schema_by_id(connection, process_schema_id)
+    service = ProcessSchemaService(connection)
+    process_schema_response = await service.get(process_schema_id)

-    if process_schema_validation is None:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Process schema not found")
+    if process_schema_response is None:
+        raise create_operation_error(
+            message="Process schema not found",
+            status_code=status.HTTP_404_NOT_FOUND,
+            details={"process_schema_id": process_schema_id},
+        )

    validator = UserRoleValidator(connection)
-    authorize_user = await validator.validate_ownership(current_user, process_schema_validation.creator_id)
+    await validator.validate_ownership(current_user, process_schema_response.process_schema.creator_id)

    updated_values = process_schema_update.model_dump(by_alias=True, exclude_none=True)

    if not updated_values:
-        return process_schema_validation
+        return process_schema_response.process_schema

-    await update_process_schema_by_id(connection, updated_values, process_schema_validation)
-
-    process_schema = await get_process_schema_by_id(connection, process_schema_id)
+    process_schema = await service.update(process_schema_id, updated_values, process_schema_response.process_schema)

    return process_schema

@@ -206,23 +146,19 @@ async def delete_process_schema_endpoint(
    connection: AsyncConnection = Depends(get_connection_dep),
    current_user=Depends(get_current_user),
 ):
-    process_schema_validation = await get_process_schema_by_id(connection, process_schema_id)
+    service = ProcessSchemaService(connection)
+    process_schema_response = await service.get(process_schema_id)

-    if process_schema_validation is None:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Process schema not found")
+    if process_schema_response is None:
+        raise create_operation_error(
+            message="Process schema not found",
+            status_code=status.HTTP_404_NOT_FOUND,
+            details={"process_schema_id": process_schema_id},
+        )

    validator = UserRoleValidator(connection)
-    authorize_user = await validator.validate_ownership(current_user, process_schema_validation.creator_id)
+    await validator.validate_ownership(current_user, process_schema_response.process_schema.creator_id)

-    process_schema_update = ProcessSchemaUpdate(status=ProcessStatus.DELETED.value)
-
-    updated_values = process_schema_update.model_dump(by_alias=True, exclude_none=True)
-
-    if not updated_values:
-        return process_schema_validation
-
-    await update_process_schema_by_id(connection, updated_values, process_schema_validation)
-
-    await get_process_schema_by_id(connection, process_schema_id)
+    await service.delete(process_schema_id, process_schema_response.process_schema)

    return HTTPException(status_code=status.HTTP_200_OK, detail="Process schema deleted successfully")
--- a/api/api/endpoints/profile.py
+++ b/api/api/endpoints/profile.py
@@ -1,17 +1,17 @@
 from fastapi import (
    APIRouter,
    Depends,
-    HTTPException,
    status,
 )
 from sqlalchemy.ext.asyncio import AsyncConnection

 from api.db.connection.session import get_connection_dep
-from api.db.logic.account import get_user_by_id, get_user_by_login, update_user_by_id
 from api.schemas.account.account import User
 from api.schemas.base import bearer_schema
 from api.schemas.endpoints.account import UserUpdate
 from api.services.auth import get_current_user
+from api.error import create_operation_error, create_validation_error
+from api.services.endpoints.profile import ProfileService

 api_router = APIRouter(
    prefix="/profile",
@@ -23,10 +23,15 @@ api_router = APIRouter(
 async def get_profile(
    connection: AsyncConnection = Depends(get_connection_dep), current_user=Depends(get_current_user)
 ):
-    user = await get_user_by_login(connection, current_user)
+    service = ProfileService(connection)
+    user = await service.get_by_login(current_user)

    if user is None:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Account not found")
+        raise create_operation_error(
+            message="Account not found",
+            status_code=status.HTTP_404_NOT_FOUND,
+            details={"user_id": current_user},
+        )

    return user

@@ -37,20 +42,25 @@ async def update_profile(
    connection: AsyncConnection = Depends(get_connection_dep),
    current_user=Depends(get_current_user),
 ):
-    user = await get_user_by_login(connection, current_user)
+    service = ProfileService(connection)
+    user = await service.get_by_login(current_user)
+
    if user is None:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Account not found")
+        raise create_operation_error(
+            message="Account not found",
+            status_code=status.HTTP_404_NOT_FOUND,
+            details={"user_id": current_user},
+        )

    if user_update.role is None and user_update.login is None:
        updated_values = user_update.model_dump(by_alias=True, exclude_none=True)

-        if updated_values is None:
-            return user
-
-        await update_user_by_id(connection, updated_values, user)
-
-        user = await get_user_by_id(connection, user.id)
+        user = await service.update(user.id, updated_values, user)

        return user
    else:
-        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Bad body")
+        raise create_validation_error(
+            message="Bad body",
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            details={"reason": "role and login fields cannot be updated"},
+        )
--- a/api/api/endpoints/ps_node.py
+++ b/api/api/endpoints/ps_node.py
@@ -4,31 +4,15 @@ from sqlalchemy.ext.asyncio import AsyncConnection

 from api.db.connection.session import get_connection_dep
 from api.db.logic.account import get_user_by_login
-
 from api.schemas.base import bearer_schema
-from api.schemas.process.process_schema import ProcessSchemaSettingsNodeLink, ProcessSchemaSettingsNode
-from api.schemas.process.ps_node import Ps_NodeFrontResponseNode, Ps_NodeRequest, Ps_NodeDeleteRequest
-from api.schemas.process.ps_node import Ps_NodeFrontResponse
+from api.schemas.process.ps_node import Ps_NodeFrontResponse, Ps_NodeRequest, Ps_NodeDeleteRequest
 from api.services.auth import get_current_user
-
-
-from api.db.logic.ps_node import (
-    create_ps_node_schema,
-    get_ps_node_by_id,
-    check_node_connection,
-    get_nodes_for_deletion_ordered,
-    delete_ps_nodes_delete_handler,
-)
-from api.db.logic.node_link import get_last_link_name_by_node_id, create_node_link_schema
-
-from api.db.logic.process_schema import update_process_schema_settings_by_id, get_process_schema_by_id
+from api.db.logic.ps_node import get_ps_node_by_id, check_node_connection
+from api.db.logic.process_schema import get_process_schema_by_id
 from api.services.user_role_validation import UserRoleValidator
-
-from core import VorkNodeRegistry, VorkNodeLink
-
-from model_nodes import VorkNodeLinkData
-from api.utils.to_camel_dict import to_camel_dict
+from core import VorkNodeRegistry
 from api.error import create_operation_error, create_access_error, create_validation_error, create_server_error
+from api.services.endpoints.ps_node import PsNodeService


 api_router = APIRouter(
@@ -77,35 +61,32 @@ async def delete_ps_node_endpoint(
            details={"next_node_id": ps_node_delete_data.next_node_id},
        )

-    is_connected = await check_node_connection(
-        connection, ps_node_delete_data.node_id, ps_node_delete_data.next_node_id, int(ps_node_delete_data.port)
-    )
+    # is_connected = await check_node_connection(
+    #     connection, ps_node_delete_data.node_id, ps_node_delete_data.next_node_id, int(ps_node_delete_data.port)
+    # )

-    if not is_connected:
-        raise create_validation_error(
-            message="Node connection validation failed",
-            status_code=status.HTTP_400_BAD_REQUEST,
-            details={
-                "node_id": ps_node_delete_data.node_id,
-                "next_node_id": ps_node_delete_data.next_node_id,
-                "port": ps_node_delete_data.port,
-            },
-        )
-
-    ordered_node_ids = await get_nodes_for_deletion_ordered(connection, ps_node_delete_data.next_node_id)
+    # if not is_connected:
+    #     raise create_validation_error(
+    #         message="Node connection validation failed",
+    #         status_code=status.HTTP_400_BAD_REQUEST,
+    #         details={
+    #             "node_id": ps_node_delete_data.node_id,
+    #             "next_node_id": ps_node_delete_data.next_node_id,
+    #             "port": ps_node_delete_data.port,
+    #         },
+    #     )

+    service = PsNodeService(connection)
    try:
-        deleted_node_ids = await delete_ps_nodes_delete_handler(connection, ordered_node_ids)
+        result = await service.delete(ps_node_delete_data.next_node_id)
    except Exception as e:
        raise create_server_error(
            message="Failed to delete nodes",
            status_code=500,
-            details={"error": str(e), "ordered_node_ids": ordered_node_ids},
+            details={"error": str(e), "next_node_id": ps_node_delete_data.next_node_id},
        )

-    return {
-        "deleted_node_ids": deleted_node_ids,
-    }
+    return result


@api_router.post("", dependencies=[Depends(bearer_schema)], response_model=Ps_NodeFrontResponse)
@@ -145,12 +126,9 @@ async def create_ps_node_endpoint(
            details={"node_type": ps_node.data["node_type"]},
        )

-    node_descriptor = vork_node.form()
    try:
        node_instance = vork_node(data=ps_node.data, links=ps_node.links)
-
        node_instance_validated = node_instance.validate()
-
    except Exception as e:
        raise create_validation_error(
            message="Node validation failed",
@@ -158,7 +136,6 @@ async def create_ps_node_endpoint(
            details={"error": str(e)},
        )

-    # Проверка: parent_id принадлежит тому же ps_id
    parent_id = node_instance_validated.parent_id
    target_ps_id = ps_node.data["ps_id"]

@@ -176,63 +153,25 @@ async def create_ps_node_endpoint(
            details={"parent_id": parent_id, "expected_ps_id": target_ps_id, "actual_ps_id": parent_node.ps_id},
        )

-    # Проверка: у родительской ноды есть указанный порт
-    parent_port_number = node_instance_validated.parent_port_number
+    # parent_port_number = node_instance_validated.parent_port_number

-    # Извлекаем номера портов из settings родительской ноды
-    parent_settings = parent_node.settings or {}
-    available_port_numbers = []
+    # parent_settings = parent_node.settings or {}
+    # available_port_numbers = []

-    # Ищем все ключи, содержащие "port" в названии
-    for key, value in parent_settings.items():
-        if "port" in key.lower() and isinstance(value, int):
-            available_port_numbers.append(value)
+    # for key, value in parent_settings.items():
+    #     if "port" in key.lower() and isinstance(value, int):
+    #         available_port_numbers.append(value)

-    # Проверяем, что указанный порт существует в settings
-    if parent_port_number not in available_port_numbers:
-        raise create_validation_error(
-            message="Parent port number is invalid",
-            status_code=status.HTTP_400_BAD_REQUEST,
-            details={"parent_id": parent_id, "parent_settings": parent_settings},
-        )
+    # if parent_port_number not in available_port_numbers:
+    #     raise create_validation_error(
+    #         message="Parent port number is invalid",
+    #         status_code=status.HTTP_400_BAD_REQUEST,
+    #         details={"parent_id": parent_id, "parent_settings": parent_settings},
+    #     )

+    service = PsNodeService(connection)
    try:
-        db_ps_node = await create_ps_node_schema(connection, node_instance_validated, user_validation.id)
-        link_name = await get_last_link_name_by_node_id(connection, db_ps_node.ps_id)
-
-        link_data = VorkNodeLinkData(
-            parent_port_number=node_instance_validated.parent_port_number,
-            to_id=db_ps_node.id,
-            from_id=node_instance_validated.parent_id,
-            last_link_name=link_name,
-        )
-
-        link = VorkNodeLink(data=link_data.model_dump())
-
-        validated_link = link.validate()
-
-        db_node_link = await create_node_link_schema(connection, validated_link, user_validation.id)
-
-        links_settings = ProcessSchemaSettingsNodeLink(
-            id=db_node_link.id,
-            link_name=db_node_link.link_name,
-            parent_port_number=db_node_link.link_point_id,
-            from_id=db_node_link.node_id,
-            to_id=db_node_link.next_node_id,
-        )
-
-        node_settings = ProcessSchemaSettingsNode(
-            id=db_ps_node.id,
-            node_type=db_ps_node.node_type,
-            data=node_instance_validated.data.model_dump(),
-            from_node=None,
-            links=[{"links": links_settings.model_dump()}],
-        )
-
-        settings_dict = {"node": node_settings.model_dump(mode="json")}
-
-        await update_process_schema_settings_by_id(connection, db_ps_node.ps_id, settings_dict)
-
+        ps_node_front_response = await service.create(ps_node.data, ps_node.links, user_validation.id)
    except Exception as e:
        raise create_server_error(
            message="Failed to create node",
@@ -240,14 +179,4 @@ async def create_ps_node_endpoint(
            details={"error": str(e)},
        )

-    ps_node_front_response = Ps_NodeFrontResponse(
-        description=node_descriptor.model_dump(),
-        node=Ps_NodeFrontResponseNode(
-            id=db_ps_node.id,
-            node_type=db_ps_node.node_type,
-            data=to_camel_dict(node_instance_validated.data.model_dump()),
-        ),
-        links=[{"links": links_settings.model_dump()}],
-    )
-
    return ps_node_front_response
--- a/api/api/schemas/process/process_schema.py
+++ b/api/api/schemas/process/process_schema.py
@@ -1,7 +1,7 @@
 from datetime import datetime
-from typing import Any, Dict, Optional, List
+from typing import Any, Dict, List

-from orm.tables.process import ProcessStatus, NodeType
+from orm.tables.process import ProcessStatus
 from pydantic import Field

 from api.schemas.base import Base
@@ -19,22 +19,6 @@ class ProcessSchema(Base):
    status: ProcessStatus


-class ProcessSchemaSettingsNodeLink(Base):
-    id: int
-    link_name: str
-    parent_port_number: int
-    from_id: int
-    to_id: int
-
-
-class ProcessSchemaSettingsNode(Base):
-    id: int
-    node_type: NodeType
-    from_node: Optional[Dict[str, Any]] = None
-    data: Dict[str, Any]  # Переименовано с 'from' на 'from_node'
-    links: Optional[List[Dict[str, Any]]] = None
-
-
 class ProcessSchemaResponse(Base):
    process_schema: ProcessSchema
-    node_listen: Ps_NodeFrontResponse
+    nodes: List[Ps_NodeFrontResponse]
--- a/api/api/schemas/process/process_version_archive.py
+++ b/api/api/schemas/process/process_version_archive.py
@@ -1,13 +0,0 @@
-from datetime import datetime
-from typing import Any, Dict
-
-from api.schemas.base import Base
-
-
-class ProcessStatusSchema(Base):
-    id: int
-    version: int
-    snapshot: Dict[str, Any]
-    owner_id: int
-    created_at: datetime
-    is_last: int
--- a/api/api/schemas/process/ps_node.py
+++ b/api/api/schemas/process/ps_node.py
@@ -28,21 +28,6 @@ class Ps_Node(Base):
    status: NodeStatus


-class Ps_NodeFrontResponseLink(Base):
-    id: int
-    link_name: str
-    parent_port_number: int
-    from_id: int
-    to_id: int
-
-
-class Ps_NodeFrontResponseNode(Base):
-    id: int
-    node_type: NodeType
-    data: Dict[str, Any]  # Переименовано с 'from' на 'from_node'
-
-
 class Ps_NodeFrontResponse(Base):
-    description: Optional[Dict[str, Any]] = None
-    node: Optional[Ps_NodeFrontResponseNode] = None
-    links: Optional[List[Dict[str, Any]]] = None
+    node: Optional[Dict[str, Any]] = None
+    link: Optional[List[Dict[str, Any]]] = None
--- a/api/api/services/endpoints/init.py
+++ b/api/api/services/endpoints/init.py
--- a/api/api/services/endpoints/account.py
+++ b/api/api/services/endpoints/account.py
@@ -0,0 +1,67 @@
+from typing import Optional
+from sqlalchemy.ext.asyncio import AsyncConnection
+from orm.tables.account import AccountStatus
+
+from api.db.logic.account import (
+    get_user_by_id,
+    update_user_by_id,
+    get_user_account_page_DTO,
+    create_user,
+)
+from api.db.logic.keyring import update_password_key, create_password_key
+from api.schemas.account.account import User
+from api.schemas.endpoints.account import AllUserResponse, UserCreate, UserFilterDTO, AllUser
+
+
+class AccountService:
+    """Сервис для работы с аккаунтами пользователей"""
+
+    def __init__(self, connection: AsyncConnection):
+        self.connection = connection
+
+    async def list(self, filter_dto: UserFilterDTO) -> Optional[AllUserResponse]:
+        """
+        Получает список пользователей с пагинацией и фильтрацией.
+        """
+        return await get_user_account_page_DTO(self.connection, filter_dto)
+
+    async def get(self, user_id: int) -> Optional[User]:
+        """
+        Получает пользователя по ID.
+        """
+        return await get_user_by_id(self.connection, user_id)
+
+    async def create(self, user_data: UserCreate, creator_id: int) -> AllUser:
+        """
+        Создаёт нового пользователя.
+        """
+        new_user = await create_user(self.connection, user_data, creator_id)
+        await create_password_key(self.connection, user_data.password, new_user.id)
+        return new_user
+
+    async def update(self, user_id: int, user_update_data: dict, password: str | None = None) -> User:
+        """
+        Обновляет данные аккаунта пользователя.
+
+        """
+        if password is not None:
+            await update_password_key(self.connection, user_id, password)
+
+        if user_update_data:
+            user = await get_user_by_id(self.connection, user_id)
+            await update_user_by_id(self.connection, user_update_data, user)
+
+        updated_user = await get_user_by_id(self.connection, user_id)
+        return updated_user
+
+    async def delete(self, user_id: int) -> User:
+        """
+        Помечает аккаунт пользователя как удалённый.
+        """
+        user = await get_user_by_id(self.connection, user_id)
+
+        update_data = {"status": AccountStatus.DELETED.value}
+        await update_user_by_id(self.connection, update_data, user)
+
+        deleted_user = await get_user_by_id(self.connection, user_id)
+        return deleted_user
--- a/api/api/services/endpoints/auth.py
+++ b/api/api/services/endpoints/auth.py
@@ -0,0 +1,47 @@
+from datetime import datetime, timedelta, timezone
+
+from fastapi_jwt_auth import AuthJWT
+from sqlalchemy.ext.asyncio import AsyncConnection
+
+from api.config import get_settings
+from api.db.logic.auth import add_new_refresh_token, upgrade_old_refresh_token
+from api.schemas.endpoints.account import AllUser
+from api.schemas.endpoints.auth import Tokens
+
+
+class AuthService:
+    """Сервис для работы с аутентификацией"""
+
+    def __init__(self, connection: AsyncConnection):
+        self.connection = connection
+
+    async def login(self, user: AllUser, authorize: AuthJWT) -> Tokens:
+        """
+        Создаёт access и refresh токены для пользователя.
+        """
+        access_token_expires = timedelta(minutes=get_settings().ACCESS_TOKEN_EXPIRE_MINUTES)
+        refresh_token_expires = timedelta(days=get_settings().REFRESH_TOKEN_EXPIRE_DAYS)
+
+        access_token = authorize.create_access_token(subject=user.login, expires_time=access_token_expires)
+        refresh_token = authorize.create_refresh_token(subject=user.login, expires_time=refresh_token_expires)
+
+        refresh_token_expires_time = datetime.now(timezone.utc) + refresh_token_expires
+
+        await add_new_refresh_token(self.connection, refresh_token, refresh_token_expires_time, user)
+
+        return Tokens(access_token=access_token, refresh_token=refresh_token)
+
+    async def refresh(self, current_user: str, authorize: AuthJWT) -> Tokens:
+        """
+        Создаёт новый access токен на основе refresh токена.
+        """
+        access_token_expires = timedelta(minutes=get_settings().ACCESS_TOKEN_EXPIRE_MINUTES)
+        new_access_token = authorize.create_access_token(subject=current_user, expires_time=access_token_expires)
+
+        return Tokens(access_token=new_access_token)
+
+    async def invalidate_refresh_token(self, refresh_token: str) -> None:
+        """
+        Помечает refresh токен как невалидный.
+        """
+        await upgrade_old_refresh_token(self.connection, refresh_token)
--- a/api/api/services/endpoints/keyring.py
+++ b/api/api/services/endpoints/keyring.py
@@ -0,0 +1,68 @@
+from typing import Optional
+from sqlalchemy.ext.asyncio import AsyncConnection
+from orm.tables.account import KeyStatus
+
+from api.db.logic.keyring import get_key_by_id, update_key_by_id, create_key
+from api.schemas.account.account_keyring import AccountKeyring
+from api.schemas.endpoints.account_keyring import AccountKeyringUpdate
+from api.error import create_validation_error
+from fastapi import status
+
+
+class KeyringService:
+    """Сервис для работы с keyring"""
+
+    def __init__(self, connection: AsyncConnection):
+        self.connection = connection
+
+    async def get(self, key_id: str) -> Optional[AccountKeyring]:
+        """
+        Получает keyring по key_id.
+        """
+        return await get_key_by_id(self.connection, key_id)
+
+    async def create(self, key: AccountKeyringUpdate, key_id: str) -> AccountKeyring:
+        """
+        Создаёт новый keyring.
+        """
+        from api.schemas.account.account_keyring import AccountKeyring
+        from datetime import datetime, timezone
+
+        if key.owner_id is None or key.key_type is None or key.key_value is None or key.status is None:
+            raise create_validation_error(
+                message="All required fields must be provided for keyring creation",
+                status_code=status.HTTP_400_BAD_REQUEST,
+            )
+
+        account_keyring = AccountKeyring(
+            owner_id=key.owner_id,
+            key_type=key.key_type,
+            key_value=key.key_value,
+            status=key.status,
+            expiry=None,
+            created_at=datetime.now(timezone.utc),
+        )
+
+        return await create_key(self.connection, account_keyring, key_id)
+
+    async def update(self, key_id: str, update_data: dict) -> AccountKeyring:
+        """
+        Обновляет данные keyring.
+        """
+        keyring = await get_key_by_id(self.connection, key_id)
+        await update_key_by_id(self.connection, update_data, keyring)
+
+        updated_keyring = await get_key_by_id(self.connection, key_id)
+        return updated_keyring
+
+    async def delete(self, key_id: str) -> AccountKeyring:
+        """
+        Помечает keyring как удалённый.
+        """
+        keyring = await get_key_by_id(self.connection, key_id)
+
+        update_data = {"status": KeyStatus.DELETED.value}
+        await update_key_by_id(self.connection, update_data, keyring)
+
+        deleted_keyring = await get_key_by_id(self.connection, key_id)
+        return deleted_keyring
--- a/api/api/services/endpoints/list_events.py
+++ b/api/api/services/endpoints/list_events.py
@@ -0,0 +1,63 @@
+from typing import Optional
+from sqlalchemy.ext.asyncio import AsyncConnection
+from orm.tables.events import EventStatus
+
+from api.db.logic.list_events import (
+    get_list_events_by_id,
+    get_list_events_by_name,
+    get_list_events_page_DTO,
+    update_list_events_by_id,
+    create_list_events,
+)
+from api.schemas.events.list_events import ListEvent
+from api.schemas.endpoints.list_events import AllListEventResponse, ListEventFilterDTO, ListEventUpdate
+
+
+class ListEventsService:
+    """Сервис для работы с list events"""
+
+    def __init__(self, connection: AsyncConnection):
+        self.connection = connection
+
+    async def list(self, filter_dto: ListEventFilterDTO) -> Optional[AllListEventResponse]:
+        """
+        Получает список событий с пагинацией и фильтрацией.
+        """
+        return await get_list_events_page_DTO(self.connection, filter_dto)
+
+    async def get(self, list_events_id: int) -> Optional[ListEvent]:
+        """
+        Получает событие по ID.
+        """
+        return await get_list_events_by_id(self.connection, list_events_id)
+
+    async def get_by_name(self, name: str) -> Optional[ListEvent]:
+        """
+        Получает событие по name.
+        """
+        return await get_list_events_by_name(self.connection, name)
+
+    async def create(self, list_events_data: ListEventUpdate, creator_id: int) -> Optional[ListEvent]:
+        """
+        Создаёт новое событие.
+        """
+
+        await create_list_events(self.connection, list_events_data, creator_id)
+        return await get_list_events_by_name(self.connection, list_events_data.name)
+
+    async def update(self, list_events_id: int, update_data: dict) -> ListEvent:
+        """
+        Обновляет данные события.
+        """
+        list_events = await get_list_events_by_id(self.connection, list_events_id)
+        await update_list_events_by_id(self.connection, update_data, list_events)
+        return await get_list_events_by_id(self.connection, list_events_id)
+
+    async def delete(self, list_events_id: int) -> ListEvent:
+        """
+        Помечает событие как удалённое.
+        """
+        list_events = await get_list_events_by_id(self.connection, list_events_id)
+        update_data = {"status": EventStatus.DELETED.value}
+        await update_list_events_by_id(self.connection, update_data, list_events)
+        return await get_list_events_by_id(self.connection, list_events_id)
--- a/api/api/services/endpoints/process_schema.py
+++ b/api/api/services/endpoints/process_schema.py
@@ -0,0 +1,131 @@
+from typing import Optional, Dict, Any
+from sqlalchemy.ext.asyncio import AsyncConnection
+from orm.tables.process import ProcessStatus
+
+from api.db.logic.process_schema import (
+    get_process_schema_by_id,
+    get_process_schema_page_DTO,
+    update_process_schema_by_id,
+    create_process_schema,
+)
+from api.db.logic.ps_node import create_ps_node_schema, get_all_ps_nodes_by_ps_id
+from api.db.logic.node_link import get_all_node_links_by_next_node_ids
+from api.schemas.process.process_schema import ProcessSchema, ProcessSchemaResponse
+from api.schemas.endpoints.process_schema import AllProcessSchemaResponse, ProcessSchemaFilterDTO, ProcessSchemaUpdate
+from api.schemas.process.ps_node import Ps_NodeFrontResponse
+from orm.tables.process import NodeType
+from core import VorkNodeRegistry
+from model_nodes import ListenNodeData
+from api.utils.node_counter import increment_node_counter
+
+
+class ProcessSchemaService:
+    """Сервис для работы с process schema"""
+
+    def __init__(self, connection: AsyncConnection):
+        self.connection = connection
+
+    async def list(self, filter_dto: ProcessSchemaFilterDTO) -> Optional[AllProcessSchemaResponse]:
+        """
+        Получает список схем процессов с пагинацией и фильтрацией.
+        """
+        return await get_process_schema_page_DTO(self.connection, filter_dto)
+
+    async def get(self, process_schema_id: int) -> Optional[ProcessSchemaResponse]:
+        """
+        Получает схему процесса по ID со всеми нодами и линками.
+        """
+        process_schema = await get_process_schema_by_id(self.connection, process_schema_id)
+        if process_schema is None:
+            return None
+
+        nodes = await get_all_ps_nodes_by_ps_id(self.connection, process_schema_id)
+
+        node_ids = [node.id for node in nodes]
+        all_links = await get_all_node_links_by_next_node_ids(self.connection, node_ids)
+
+        links_by_node_id = {}
+        for link in all_links:
+            if link.next_node_id not in links_by_node_id:
+                links_by_node_id[link.next_node_id] = []
+            links_by_node_id[link.next_node_id].append(link)
+
+        nodes_response = []
+        for node in nodes:
+            node_links = links_by_node_id.get(node.id, [])
+            links_list = [{"link": link.model_dump()} for link in node_links]
+
+            ps_node_front_response = Ps_NodeFrontResponse(
+                node=node.model_dump(),
+                link=links_list,
+            )
+            nodes_response.append(ps_node_front_response)
+
+        return ProcessSchemaResponse(
+            process_schema=process_schema,
+            nodes=nodes_response,
+        )
+
+    async def create(self, creator_id: int) -> Dict[str, Any]:
+        """
+        Создаёт новую схему процесса с начальной нодой LISTEN.
+        """
+        current_node_counter = increment_node_counter()
+        title = f"Новая схема {current_node_counter}"
+        description = "Default description"
+
+        node_id = await create_process_schema(self.connection, creator_id, title, description)
+
+        process_schema_new = await get_process_schema_by_id(self.connection, node_id)
+
+        start_node_data = ListenNodeData(ps_id=process_schema_new.id, node_type=NodeType.START.value, is_start="True")
+
+        start_node_links = {}
+
+        registery = VorkNodeRegistry()
+
+        vork_node = registery.get("LISTEN")
+
+        node_descriptor = vork_node.form()
+
+        start_node = vork_node(data=start_node_data.model_dump(), links=start_node_links)
+
+        validated_start_schema = start_node.validate()
+
+        start_settings_payload = {
+            **node_descriptor.model_dump(),
+            **validated_start_schema.data.model_dump(),
+        }
+
+        db_start_schema = await create_ps_node_schema(
+            self.connection,
+            validated_start_schema,
+            node_descriptor,
+            creator_id,
+            start_settings_payload,
+        )
+
+        ps_node_front_response = Ps_NodeFrontResponse(
+            node=db_start_schema.model_dump(),
+            link=[],
+        )
+        response_data = {
+            "process_schema": process_schema_new.model_dump(),
+            "nodes": [ps_node_front_response],  # Список объектов, а не словарей
+        }
+        return response_data
+
+    async def update(self, process_schema_id: int, update_data: dict, process_schema: ProcessSchema) -> ProcessSchema:
+        """
+        Обновляет данные схемы процесса.
+        """
+        await update_process_schema_by_id(self.connection, update_data, process_schema)
+        return await get_process_schema_by_id(self.connection, process_schema_id)
+
+    async def delete(self, process_schema_id: int, process_schema: ProcessSchema) -> None:
+        """
+        Помечает схему процесса как удалённую.
+        """
+        process_schema_update = ProcessSchemaUpdate(status=ProcessStatus.DELETED.value)
+        update_data = process_schema_update.model_dump(by_alias=True, exclude_none=True)
+        await update_process_schema_by_id(self.connection, update_data, process_schema)
--- a/api/api/services/endpoints/profile.py
+++ b/api/api/services/endpoints/profile.py
@@ -0,0 +1,30 @@
+from typing import Optional
+from sqlalchemy.ext.asyncio import AsyncConnection
+
+from api.db.logic.account import get_user_by_id, get_user_by_login, update_user_by_id
+from api.schemas.account.account import User
+
+
+class ProfileService:
+    """Сервис для работы с профилем пользователя"""
+
+    def __init__(self, connection: AsyncConnection):
+        self.connection = connection
+
+    async def get_by_login(self, login: str) -> Optional[User]:
+        """
+        Получает пользователя по логину.
+        """
+        return await get_user_by_login(self.connection, login)
+
+    async def update(self, user_id: int, update_data: dict, user: User) -> User:
+        """
+        Обновляет данные профиля пользователя.
+        """
+        if update_data is None or not update_data:
+            return user
+
+        await update_user_by_id(self.connection, update_data, user)
+
+        updated_user = await get_user_by_id(self.connection, user_id)
+        return updated_user
--- a/api/api/services/endpoints/ps_node.py
+++ b/api/api/services/endpoints/ps_node.py
@@ -0,0 +1,65 @@
+from typing import List, Dict, Any
+from sqlalchemy.ext.asyncio import AsyncConnection
+
+from api.db.logic.ps_node import (
+    get_nodes_for_deletion_ordered,
+    delete_ps_nodes_delete_handler,
+    create_ps_node_schema,
+)
+from api.db.logic.node_link import get_last_link_name_by_node_id, create_node_link_schema
+from api.schemas.process.ps_node import Ps_NodeFrontResponse
+from core import VorkNodeRegistry, VorkNodeLink
+from model_nodes import VorkNodeLinkData
+
+
+class PsNodeService:
+    """Сервис для работы с ps nodes"""
+
+    def __init__(self, connection: AsyncConnection):
+        self.connection = connection
+
+    async def delete(self, next_node_id: int) -> Dict[str, List[int]]:
+        """
+        Удаляет ноды в правильном порядке.
+        """
+        ordered_node_ids = await get_nodes_for_deletion_ordered(self.connection, next_node_id)
+        deleted_node_ids = await delete_ps_nodes_delete_handler(self.connection, ordered_node_ids)
+
+        return {
+            "deleted_node_ids": deleted_node_ids,
+        }
+
+    async def create(
+        self, ps_node_data: Dict[str, Any], links: Dict[str, Any], creator_id: int
+    ) -> Ps_NodeFrontResponse:
+        """
+        Создаёт новую ноду с линком и обновляет настройки схемы процесса.
+        """
+        registery = VorkNodeRegistry()
+        vork_node = registery.get(ps_node_data["node_type"])
+        node_descriptor = vork_node.form()
+
+        node_instance = vork_node(data=ps_node_data, links=links)
+        node_instance_validated = node_instance.validate()
+
+        db_ps_node = await create_ps_node_schema(self.connection, node_instance_validated, node_descriptor, creator_id)
+        link_name = await get_last_link_name_by_node_id(self.connection, db_ps_node.ps_id)
+
+        link_data = VorkNodeLinkData(
+            parent_port_number=node_instance_validated.parent_port_number,
+            to_id=db_ps_node.id,
+            from_id=node_instance_validated.parent_id,
+            last_link_name=link_name,
+        )
+
+        link = VorkNodeLink(data=link_data.model_dump())
+        validated_link = link.validate()
+
+        db_node_link = await create_node_link_schema(self.connection, validated_link, creator_id)
+
+        ps_node_front_response = Ps_NodeFrontResponse(
+            node=db_ps_node.model_dump(),
+            link=[{"link": db_node_link.model_dump()}],
+        )
+
+        return ps_node_front_response
--- a/api/api/services/user_role_validation.py
+++ b/api/api/services/user_role_validation.py
@@ -11,7 +11,7 @@ class UserRoleValidator:
        self.connection = connection

    async def validate_admin(self, current_user: int):
-        """Проверяет права администратора или владельца"""
+        """Проверяет права пользователя"""
        try:
            authorize_user = await get_user_by_login(self.connection, current_user)
        except Exception as e:
Author	SHA1	Message	Date
TheNoxium	0330356ea2	feat: settings, delete port validation	2025-11-15 16:00:05 +05:00
ivan.dev	46091380ba	Merge pull request 'feat: endpoints services' (#21 ) from codestyle-issue-refactor into master Reviewed-on: #21 Reviewed-by: Vladislav Syrochkin <vlad.dev@heado.ru> Reviewed-by: cyrussmeat <dr.cyrill@gmail.com>	2025-11-11 18:05:30 +05:00
TheNoxium	3933a05582	feat: endpoints services	2025-11-06 23:34:47 +05:00
ivan.dev	453efdd095	Merge pull request 'feat: delete ps node' (#20 ) from VORKOUT-29 into master Reviewed-on: #20 Reviewed-by: cyrussmeat <dr.cyrill@gmail.com> Reviewed-by: Vladislav Syrochkin <vlad.dev@heado.ru>	2025-11-05 14:28:51 +05:00