Param alias

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
 

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')

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):

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
 
@@ -41,5 +42,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)

userdata_api/routes/param_alias.py

@@ -0,0 +1,141 @@
+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="/category/{category_id}/param/{param_id}/alias", tags=["Param Alias"])
+
+
+def _get_param(*, category_id: int, param_id: int) -> Param:
+    param = (
+        Param.query(session=db.session)
+        .filter(
+            Param.id == param_id,
+            Param.category_id == category_id,
+        )
+        .one_or_none()
+    )
+    if not param:
+        raise ObjectNotFound(Param, param_id)
+    return param
+
+
+def _get_param_alias(*, category_id: int, 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,
+            Param.category_id == category_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,
+    category_id: int,
+    param_id: int,
+    alias_inp: ParamAliasPost,
+    _: dict[str, Any] = Depends(UnionAuth(scopes=["userdata.param.create"], allow_none=False, auto_error=True)),
+) -> ParamAliasGet:
+    """
+    Создать алиас параметра.
+    """
+    _ = _get_param(category_id=category_id, 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(
+    category_id: int,
+    param_id: int,
+    alias_id: int,
+) -> ParamAliasGet:
+    """
+    Получить алиас параметра по айди.
+    """
+    alias = _get_param_alias(category_id=category_id, 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(category_id: int, param_id: int) -> list[ParamAliasGet]:
+    """
+    Получить все алиасы параметра.
+    """
+    _ = _get_param(category_id=category_id, 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(
+    category_id: int,
+    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(category_id=category_id, 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,
+    category_id: int,
+    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(category_id=category_id, param_id=param_id, alias_id=alias_id)
+    ParamAlias.delete(alias_id, session=db.session)
+    return StatusResponseModel(status="Success", message="Param alias deleted", ru="Алиас параметра удален")

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

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

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()

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)