diff --git a/Makefile b/Makefile index 2c792ab..cb70bc7 100644 --- a/Makefile +++ b/Makefile @@ -1,3 +1,5 @@ +SHELL := /bin/bash + run: source ./venv/bin/activate && uvicorn --reload --log-config logging_dev.conf userdata_api.routes.base:app diff --git a/migrations/versions/8b49f52e2c01_add_param_alias_table.py b/migrations/versions/8b49f52e2c01_add_param_alias_table.py new file mode 100644 index 0000000..0353746 --- /dev/null +++ b/migrations/versions/8b49f52e2c01_add_param_alias_table.py @@ -0,0 +1,44 @@ +"""Add param_alias table + +Revision ID: 8b49f52e2c01 +Revises: fc911d58459b +Create Date: 2026-05-16 12:10:00.000000 + +""" + +import sqlalchemy as sa +from alembic import op + + +# revision identifiers, used by Alembic. +revision = '8b49f52e2c01' +down_revision = 'fc911d58459b' +branch_labels = None +depends_on = None + + +def upgrade(): + op.create_table( + 'param_alias', + sa.Column('name', sa.String(), nullable=False), + sa.Column('param_id', sa.Integer(), nullable=False), + sa.Column('source_id', sa.Integer(), nullable=True), + sa.Column('create_ts', sa.DateTime(), nullable=False), + sa.Column('modify_ts', sa.DateTime(), nullable=False), + sa.Column('is_deleted', sa.Boolean(), nullable=False), + sa.Column('id', sa.Integer(), nullable=False), + sa.ForeignKeyConstraint( + ['param_id'], + ['param.id'], + ), + sa.ForeignKeyConstraint( + ['source_id'], + ['source.id'], + ), + sa.PrimaryKeyConstraint('id'), + sa.UniqueConstraint('name'), + ) + + +def downgrade(): + op.drop_table('param_alias') diff --git a/tests/test_routes/test_param_alias.py b/tests/test_routes/test_param_alias.py new file mode 100644 index 0000000..b5fee6e --- /dev/null +++ b/tests/test_routes/test_param_alias.py @@ -0,0 +1,134 @@ +import pytest + +from userdata_api.models.db import ParamAlias +from userdata_api.utils.utils import random_string + + +@pytest.mark.authenticated("userdata.param.create") +def test_create_param_alias(client, dbsession, param, source): + _param = param() + _source = source() + alias_name = f"alias_{random_string()}" + response = client.post( + f"/param/{_param.id}/alias", + json={"name": alias_name, "source_id": _source.id}, + ) + assert response.status_code == 200 + assert response.json()["name"] == alias_name + assert response.json()["param_id"] == _param.id + assert response.json()["source_id"] == _source.id + assert response.json()["source_name"] == _source.name + alias = ParamAlias.get(response.json()["id"], session=dbsession) + assert alias + assert alias.name == alias_name + dbsession.delete(alias) + dbsession.commit() + + +@pytest.mark.authenticated("userdata.param.create") +def test_create_param_alias_global(client, dbsession, param): + _param = param() + alias_name = f"alias_{random_string()}" + response = client.post( + f"/param/{_param.id}/alias", + json={"name": alias_name}, + ) + assert response.status_code == 200 + assert response.json()["name"] == alias_name + assert response.json()["param_id"] == _param.id + assert response.json()["source_id"] is None + assert response.json()["source_name"] is None + alias = ParamAlias.get(response.json()["id"], session=dbsession) + dbsession.delete(alias) + dbsession.commit() + + +@pytest.mark.authenticated("userdata.param.create") +def test_create_param_alias_unique_whole_table(client, dbsession, param): + _param1 = param() + _param2 = param() + alias_name = f"alias_{random_string()}" + response_1 = client.post( + f"/param/{_param1.id}/alias", + json={"name": alias_name}, + ) + assert response_1.status_code == 200 + response_2 = client.post( + f"/param/{_param2.id}/alias", + json={"name": alias_name}, + ) + assert response_2.status_code == 409 + alias = ParamAlias.get(response_1.json()["id"], session=dbsession) + dbsession.delete(alias) + dbsession.commit() + + +@pytest.mark.authenticated("userdata.param.create") +def test_get_param_aliases(client, dbsession, param, source): + _param = param() + _source = source() + first_name = f"alias_{random_string()}" + second_name = f"alias_{random_string()}" + create_1 = client.post( + f"/param/{_param.id}/alias", + json={"name": first_name, "source_id": _source.id}, + ) + create_2 = client.post( + f"/param/{_param.id}/alias", + json={"name": second_name}, + ) + assert create_1.status_code == 200 + assert create_2.status_code == 200 + response = client.get(f"/param/{_param.id}/alias") + assert response.status_code == 200 + assert any(item["name"] == first_name and item["source_id"] == _source.id for item in response.json()) + assert any(item["name"] == second_name and item["source_id"] is None for item in response.json()) + dbsession.delete(ParamAlias.get(create_1.json()["id"], session=dbsession)) + dbsession.delete(ParamAlias.get(create_2.json()["id"], session=dbsession)) + dbsession.commit() + + +@pytest.mark.authenticated("userdata.param.create", "userdata.param.update") +def test_patch_param_alias(client, dbsession, param, source): + _param = param() + _source = source() + _new_source = source() + alias_name = f"alias_{random_string()}" + alias_new_name = f"alias_{random_string()}" + create = client.post( + f"/param/{_param.id}/alias", + json={"name": alias_name, "source_id": _source.id}, + ) + assert create.status_code == 200 + alias_id = create.json()["id"] + response = client.patch( + f"/param/{_param.id}/alias/{alias_id}", + json={"name": alias_new_name, "source_id": _new_source.id}, + ) + assert response.status_code == 200 + assert response.json()["name"] == alias_new_name + assert response.json()["source_id"] == _new_source.id + dbsession.expire_all() + alias = ParamAlias.get(alias_id, session=dbsession) + assert alias.name == alias_new_name + assert alias.source_id == _new_source.id + dbsession.delete(alias) + dbsession.commit() + + +@pytest.mark.authenticated("userdata.param.create", "userdata.param.delete") +def test_delete_param_alias(client, dbsession, param): + _param = param() + alias_name = f"alias_{random_string()}" + create = client.post( + f"/param/{_param.id}/alias", + json={"name": alias_name}, + ) + assert create.status_code == 200 + alias_id = create.json()["id"] + response = client.delete(f"/param/{_param.id}/alias/{alias_id}") + assert response.status_code == 200 + alias = ParamAlias.get(alias_id, session=dbsession, with_deleted=True) + assert alias.is_deleted is True + dbsession.delete(alias) + dbsession.commit() diff --git a/tests/test_routes/test_user_update.py b/tests/test_routes/test_user_update.py index 8b1e8e2..cbf29f1 100644 --- a/tests/test_routes/test_user_update.py +++ b/tests/test_routes/test_user_update.py @@ -440,3 +440,92 @@ def test_delete_forbidden_by_category_scope(dbsession, client, param, admin_sour assert response.status_code == 403 assert not info1.is_deleted dbsession.delete(info1) + + +@pytest.mark.authenticated("test.cat_update.first", "userdata.info.admin", user_id=1) +def test_update_by_source_alias(dbsession, client, param, admin_source): + _param = param() + _param.category.update_scope = "test.cat_update.first" + alias_name = f"alias_{random_string()}" + dbsession.add(ParamAlias(name=alias_name, param_id=_param.id, source_id=admin_source.id)) + dbsession.commit() + response = client.post( + f"/user/0", + json={ + "source": "admin", + "items": [{"category": _param.category.name, "param": alias_name, "value": "first_updated"}], + }, + ) + dbsession.expire_all() + assert response.status_code == 200 + info = ( + dbsession.query(Info) + .filter( + Info.param_id == _param.id, Info.owner_id == 0, Info.source_id == admin_source.id, Info.is_deleted == False + ) + .one() + ) + assert info.value == "first_updated" + alias = dbsession.query(ParamAlias).filter(ParamAlias.name == alias_name).one() + dbsession.delete(info) + dbsession.delete(alias) + dbsession.commit() + + +@pytest.mark.authenticated("test.cat_update.first", "userdata.info.admin", user_id=1) +def test_update_by_global_alias(dbsession, client, param, admin_source): + _param = param() + _param.category.update_scope = "test.cat_update.first" + alias_name = f"alias_{random_string()}" + dbsession.add(ParamAlias(name=alias_name, param_id=_param.id, source_id=None)) + dbsession.commit() + response = client.post( + f"/user/0", + json={ + "source": "admin", + "items": [{"category": _param.category.name, "param": alias_name, "value": "global_updated"}], + }, + ) + dbsession.expire_all() + assert response.status_code == 200 + info = ( + dbsession.query(Info) + .filter( + Info.param_id == _param.id, Info.owner_id == 0, Info.source_id == admin_source.id, Info.is_deleted == False + ) + .one() + ) + assert info.value == "global_updated" + alias = dbsession.query(ParamAlias).filter(ParamAlias.name == alias_name).one() + dbsession.delete(info) + dbsession.delete(alias) + dbsession.commit() + + +@pytest.mark.authenticated("test.cat_update.first", "userdata.info.admin", user_id=1) +def test_update_by_foreign_source_alias_not_found(dbsession, client, param, admin_source, source): + _param = param() + _param.category.update_scope = "test.cat_update.first" + _source = source() + alias_name = f"alias_{random_string()}" + dbsession.add(ParamAlias(name=alias_name, param_id=_param.id, source_id=_source.id)) + dbsession.commit() + response = client.post( + f"/user/0", + json={ + "source": "admin", + "items": [{"category": _param.category.name, "param": alias_name, "value": "should_not_work"}], + }, + ) + assert response.status_code == 404 + info = ( + dbsession.query(Info) + .filter( + Info.param_id == _param.id, Info.owner_id == 0, Info.source_id == admin_source.id, Info.is_deleted == False + ) + .one_or_none() + ) + assert info is None + alias = dbsession.query(ParamAlias).filter(ParamAlias.name == alias_name).one() + dbsession.delete(alias) + dbsession.commit() diff --git a/tests/test_worker/test_worker.py b/tests/test_worker/test_worker.py index e1dd4a5..fb705a3 100644 --- a/tests/test_worker/test_worker.py +++ b/tests/test_worker/test_worker.py @@ -2,7 +2,7 @@ import sqlalchemy.exc from event_schema.auth import UserLogin -from userdata_api.models.db import Category, Info, Param, Source +from userdata_api.models.db import Category, Info, Param, ParamAlias, Source from userdata_api.utils.utils import random_string from worker.user import patch_user_info @@ -108,3 +108,88 @@ def test_delete(info, dbsession): dbsession.expire(info) assert info.is_deleted is True + + +def test_create_by_alias(param, source, dbsession): + alias_name = f"alias_{random_string()}" + alias = ParamAlias(name=alias_name, param_id=param.id, source_id=source.id) + dbsession.add(alias) + dbsession.commit() + patch_user_info( + UserLogin.model_validate( + { + "items": [{"category": param.category.name, "param": alias_name, "value": "test_by_alias"}], + "source": source.name, + } + ), + 1, + session=dbsession, + ) + info = ( + dbsession.query(Info) + .filter( + Info.param_id == param.id, Info.source_id == source.id, Info.owner_id == 1, Info.value == "test_by_alias" + ) + .one() + ) + assert info + dbsession.delete(info) + dbsession.delete(alias) + dbsession.commit() + + +def test_create_by_global_alias(param, source, dbsession): + alias_name = f"alias_{random_string()}" + alias = ParamAlias(name=alias_name, param_id=param.id, source_id=None) + dbsession.add(alias) + dbsession.commit() + patch_user_info( + UserLogin.model_validate( + { + "items": [{"category": param.category.name, "param": alias_name, "value": "test_by_global_alias"}], + "source": source.name, + } + ), + 1, + session=dbsession, + ) + info = ( + dbsession.query(Info) + .filter( + Info.param_id == param.id, + Info.source_id == source.id, + Info.owner_id == 1, + Info.value == "test_by_global_alias", + ) + .one() + ) + assert info + dbsession.delete(info) + dbsession.delete(alias) + dbsession.commit() + + +def test_create_by_foreign_source_alias_not_found(param, source, dbsession): + first_source = source + second_source = Source(name=f"test{random_string()}", trust_level=8) + dbsession.add(second_source) + dbsession.commit() + alias_name = f"alias_{random_string()}" + alias = ParamAlias(name=alias_name, param_id=param.id, source_id=second_source.id) + dbsession.add(alias) + dbsession.commit() + patch_user_info( + UserLogin.model_validate( + { + "items": [{"category": param.category.name, "param": alias_name, "value": "should_not_work"}], + "source": first_source.name, + } + ), + 1, + session=dbsession, + ) + with pytest.raises(sqlalchemy.exc.NoResultFound): + dbsession.query(Info).filter(Info.param_id == param.id, Info.value == "should_not_work").one() + dbsession.delete(alias) + dbsession.delete(second_source) + dbsession.commit() diff --git a/userdata_api/models/db.py b/userdata_api/models/db.py index 2ba3da5..aa5b5ad 100644 --- a/userdata_api/models/db.py +++ b/userdata_api/models/db.py @@ -84,6 +84,13 @@ class Param(BaseDbModel): primaryjoin="and_(Param.id==Info.param_id, not_(Info.is_deleted))", lazy="joined", ) + aliases: Mapped[list[ParamAlias]] = relationship( + "ParamAlias", + foreign_keys="ParamAlias.param_id", + back_populates="param", + primaryjoin="and_(Param.id==ParamAlias.param_id, not_(ParamAlias.is_deleted))", + lazy="joined", + ) @property def pytype(self) -> type[str | list[str]]: @@ -111,6 +118,47 @@ class Source(BaseDbModel): primaryjoin="and_(Source.id==Info.source_id, not_(Info.is_deleted))", lazy="joined", ) + aliases: Mapped[list[ParamAlias]] = relationship( + "ParamAlias", + foreign_keys="ParamAlias.source_id", + back_populates="source", + primaryjoin="and_(Source.id==ParamAlias.source_id, not_(ParamAlias.is_deleted))", + lazy="joined", + ) + + +class ParamAlias(BaseDbModel): + """ + Алиас параметра. + Может быть привязан к конкретному источнику или быть общим для всех источников. + """ + + name: Mapped[str] = mapped_column(String, unique=True) + param_id: Mapped[int] = mapped_column(Integer, ForeignKey(Param.id)) + source_id: Mapped[int | None] = mapped_column(Integer, ForeignKey(Source.id), nullable=True) + create_ts: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) + modify_ts: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) + is_deleted: Mapped[bool] = mapped_column(Boolean, default=False) + + param: Mapped[Param] = relationship( + "Param", + foreign_keys="ParamAlias.param_id", + back_populates="aliases", + primaryjoin="and_(ParamAlias.param_id==Param.id, not_(Param.is_deleted))", + lazy="joined", + ) + + source: Mapped[Source | None] = relationship( + "Source", + foreign_keys="ParamAlias.source_id", + back_populates="aliases", + primaryjoin="and_(ParamAlias.source_id==Source.id, not_(Source.is_deleted))", + lazy="joined", + ) + + @hybrid_property + def source_name(self) -> str | None: + return self.source.name if self.source else None class Info(BaseDbModel): diff --git a/userdata_api/routes/base.py b/userdata_api/routes/base.py index 9e90730..8e6a0e5 100644 --- a/userdata_api/routes/base.py +++ b/userdata_api/routes/base.py @@ -8,6 +8,7 @@ from .admin import admin from .category import category from .param import param +from .param_alias import param_alias from .source import source from .user import user @@ -40,5 +41,6 @@ app.include_router(source) app.include_router(category) app.include_router(param) +app.include_router(param_alias) app.include_router(user) app.include_router(admin) diff --git a/userdata_api/routes/param_alias.py b/userdata_api/routes/param_alias.py new file mode 100644 index 0000000..d44bf84 --- /dev/null +++ b/userdata_api/routes/param_alias.py @@ -0,0 +1,129 @@ +from typing import Any + +from auth_lib.fastapi import UnionAuth +from fastapi import APIRouter, Depends, Request +from fastapi_sqlalchemy import db +from pydantic.type_adapter import TypeAdapter +from sqlalchemy import not_ + +from userdata_api.exceptions import AlreadyExists, ObjectNotFound +from userdata_api.models.db import Param, ParamAlias, Source +from userdata_api.schemas.param_alias import ParamAliasGet, ParamAliasPatch, ParamAliasPost +from userdata_api.schemas.response_model import StatusResponseModel + + +param_alias = APIRouter(prefix="/param/{param_id}/alias", tags=["Param Alias"]) + + +def _get_param(*, param_id: int) -> Param: + param = Param.query(session=db.session).filter(Param.id == param_id).one_or_none() + if not param: + raise ObjectNotFound(Param, param_id) + return param + + +def _get_param_alias(*, param_id: int, alias_id: int) -> ParamAlias: + alias = ( + ParamAlias.query(session=db.session) + .join(Param) + .filter( + ParamAlias.id == alias_id, + ParamAlias.param_id == param_id, + not_(Param.is_deleted), + ) + .one_or_none() + ) + if not alias: + raise ObjectNotFound(ParamAlias, alias_id) + return alias + + +def _validate_source(source_id: int | None) -> None: + if source_id is None: + return + source = Source.query(session=db.session).filter(Source.id == source_id).one_or_none() + if not source: + raise ObjectNotFound(Source, source_id) + + +def _check_alias_name_exists(name: str, *, alias_id: int | None = None) -> None: + query = db.session.query(ParamAlias).filter(ParamAlias.name == name) + if alias_id is not None: + query = query.filter(ParamAlias.id != alias_id) + if query.one_or_none(): + raise AlreadyExists(ParamAlias, name) + + +@param_alias.post("", response_model=ParamAliasGet) +async def create_param_alias( + request: Request, + param_id: int, + alias_inp: ParamAliasPost, + _: dict[str, Any] = Depends(UnionAuth(scopes=["userdata.param.create"], allow_none=False, auto_error=True)), +) -> ParamAliasGet: + """ + Создать алиас параметра. + """ + _ = _get_param(param_id=param_id) + _validate_source(alias_inp.source_id) + _check_alias_name_exists(alias_inp.name) + alias = ParamAlias.create(session=db.session, param_id=param_id, **alias_inp.model_dump()) + return ParamAliasGet.model_validate(alias) + + +@param_alias.get("/{alias_id}", response_model=ParamAliasGet) +async def get_param_alias( + param_id: int, + alias_id: int, +) -> ParamAliasGet: + """ + Получить алиас параметра по айди. + """ + alias = _get_param_alias(param_id=param_id, alias_id=alias_id) + return ParamAliasGet.model_validate(alias) + + +@param_alias.get("", response_model=list[ParamAliasGet]) +async def get_param_aliases(param_id: int) -> list[ParamAliasGet]: + """ + Получить все алиасы параметра. + """ + _ = _get_param(param_id=param_id) + aliases = ParamAlias.query(session=db.session).filter(ParamAlias.param_id == param_id).all() + type_adapter = TypeAdapter(list[ParamAliasGet]) + return type_adapter.validate_python(aliases) + + +@param_alias.patch("/{alias_id}", response_model=ParamAliasGet) +async def patch_param_alias( + param_id: int, + alias_id: int, + alias_inp: ParamAliasPatch, + _: dict[str, Any] = Depends(UnionAuth(scopes=["userdata.param.update"], allow_none=False, auto_error=True)), +) -> ParamAliasGet: + """ + Обновить алиас параметра. + """ + alias = _get_param_alias(param_id=param_id, alias_id=alias_id) + patch_data = alias_inp.model_dump(exclude_unset=True) + if "name" in patch_data: + _check_alias_name_exists(patch_data["name"], alias_id=alias.id) + if "source_id" in patch_data: + _validate_source(patch_data["source_id"]) + alias = ParamAlias.update(alias.id, session=db.session, **patch_data) + return ParamAliasGet.model_validate(alias) + + +@param_alias.delete("/{alias_id}", response_model=StatusResponseModel) +async def delete_param_alias( + request: Request, + param_id: int, + alias_id: int, + _: dict[str, Any] = Depends(UnionAuth(scopes=["userdata.param.delete"], allow_none=False, auto_error=True)), +) -> StatusResponseModel: + """ + Удалить алиас параметра. + """ + _ = _get_param_alias(param_id=param_id, alias_id=alias_id) + ParamAlias.delete(alias_id, session=db.session) + return StatusResponseModel(status="Success", message="Param alias deleted", ru="Алиас параметра удален") diff --git a/userdata_api/schemas/param.py b/userdata_api/schemas/param.py index 7a4bc01..3c0e3ce 100644 --- a/userdata_api/schemas/param.py +++ b/userdata_api/schemas/param.py @@ -3,6 +3,7 @@ from userdata_api.models.db import ViewType from .base import Base +from .param_alias import ParamAliasGet class ParamPost(Base): @@ -28,3 +29,4 @@ class ParamPatch(Base): class ParamGet(ParamPost): id: int category_id: int + aliases: list[ParamAliasGet] | None = None diff --git a/userdata_api/schemas/param_alias.py b/userdata_api/schemas/param_alias.py new file mode 100644 index 0000000..b580e5e --- /dev/null +++ b/userdata_api/schemas/param_alias.py @@ -0,0 +1,19 @@ +from pydantic import constr + +from .base import Base + + +class ParamAliasPost(Base): + name: constr(min_length=1) + source_id: int | None = None + + +class ParamAliasPatch(Base): + name: constr(min_length=1) | None = None + source_id: int | None = None + + +class ParamAliasGet(ParamAliasPost): + id: int + param_id: int + source_name: str | None = None diff --git a/userdata_api/utils/param_alias.py b/userdata_api/utils/param_alias.py new file mode 100644 index 0000000..65b2f7a --- /dev/null +++ b/userdata_api/utils/param_alias.py @@ -0,0 +1,52 @@ +from sqlalchemy import not_, or_ +from sqlalchemy.orm import Session + +from userdata_api.models.db import Category, Param, ParamAlias, Source + + +def get_param_by_name_or_alias( + *, + session: Session, + category_name: str, + param_name: str, + source_name: str, +) -> Param | None: + """ + Находит параметр сначала по каноническому имени, затем по алиасу. + + Если у алиаса source_id = NULL, алиас считается общим для всех источников. + """ + param = ( + session.query(Param) + .join(Category) + .filter( + Param.name == param_name, + Category.name == category_name, + not_(Param.is_deleted), + not_(Category.is_deleted), + ) + .one_or_none() + ) + if param: + return param + + source = Source.query(session=session).filter(Source.name == source_name).one_or_none() + source_id = source.id if source else None + + query = ( + session.query(Param) + .join(Category) + .join(ParamAlias, ParamAlias.param_id == Param.id) + .filter( + ParamAlias.name == param_name, + Category.name == category_name, + not_(ParamAlias.is_deleted), + not_(Param.is_deleted), + not_(Category.is_deleted), + ) + ) + if source_id is None: + query = query.filter(ParamAlias.source_id.is_(None)) + else: + query = query.filter(or_(ParamAlias.source_id == source_id, ParamAlias.source_id.is_(None))) + return query.one_or_none() diff --git a/userdata_api/utils/user.py b/userdata_api/utils/user.py index a854e00..882aef2 100644 --- a/userdata_api/utils/user.py +++ b/userdata_api/utils/user.py @@ -8,6 +8,7 @@ from userdata_api.exceptions import Forbidden, InvalidValidation, ObjectNotFound from userdata_api.models.db import Category, Info, Param, Source, ViewType from userdata_api.schemas.user import UserInfoGet, UserInfoUpdate, UsersInfoGet +from userdata_api.utils.param_alias import get_param_by_name_or_alias async def patch_user_info(new: UserInfoUpdate, user_id: int, user: dict[str, int | list[dict[str, str | int]]]) -> None: @@ -50,16 +51,11 @@ async def patch_user_info(new: UserInfoUpdate, user_id: int, user: dict[str, int if new.source == "user" and user["id"] != user_id: raise Forbidden("'user' source requires information own", "Требуется владение информацией") for item in new.items: - param = ( - db.session.query(Param) - .join(Category) - .filter( - Param.name == item.param, - Category.name == item.category, - not_(Param.is_deleted), - not_(Category.is_deleted), - ) - .one_or_none() + param = get_param_by_name_or_alias( + session=db.session, + category_name=item.category, + param_name=item.param, + source_name=new.source, ) if not param: raise ObjectNotFound(Param, item.param) diff --git a/worker/user.py b/worker/user.py index e9b0371..664b60a 100644 --- a/worker/user.py +++ b/worker/user.py @@ -4,23 +4,19 @@ from event_schema.auth import UserLogin from sqlalchemy import not_ -from userdata_api.models.db import Category, Info, Param, Source +from userdata_api.models.db import Info, Source +from userdata_api.utils.param_alias import get_param_by_name_or_alias log = logging.getLogger(__name__) def patch_user_info(new: UserLogin, user_id: int, *, session: sqlalchemy.orm.Session) -> None: for item in new.items: - param = ( - session.query(Param) - .join(Category) - .filter( - Param.name == item.param, - Category.name == item.category, - not_(Param.is_deleted), - not_(Category.is_deleted), - ) - .one_or_none() + param = get_param_by_name_or_alias( + session=session, + category_name=item.category, + param_name=item.param, + source_name=new.source, ) if not param: session.rollback()