802 add support for automatic foulborn unique tracking

src/backend_api/app/alembic/versions/d789cc84cb4d_created_carantene_modifier_table.py

@@ -0,0 +1,57 @@
+"""Created carantene modifier table
+
+Revision ID: d789cc84cb4d
+Revises: e38727349f3f
+Create Date: 2025-11-18 19:44:38.570115
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'd789cc84cb4d'
+down_revision: Union[str, None] = 'e38727349f3f'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('carantene_modifier',
+    sa.Column('caranteneModifierId', sa.BigInteger(), sa.Identity(always=False, start=1, increment=1, cycle=True), nullable=False),
+    sa.Column('effect', sa.Text(), nullable=False),
+    sa.Column('relatedUnique', sa.Text(), nullable=False),
+    sa.Column('implicit', sa.Boolean(), nullable=True),
+    sa.Column('explicit', sa.Boolean(), nullable=True),
+    sa.Column('delve', sa.Boolean(), nullable=True),
+    sa.Column('fractured', sa.Boolean(), nullable=True),
+    sa.Column('synthesised', sa.Boolean(), nullable=True),
+    sa.Column('unique', sa.Boolean(), nullable=True),
+    sa.Column('corrupted', sa.Boolean(), nullable=True),
+    sa.Column('enchanted', sa.Boolean(), nullable=True),
+    sa.Column('veiled', sa.Boolean(), nullable=True),
+    sa.Column('mutated', sa.Boolean(), nullable=True),
+    sa.Column('createdAt', sa.DateTime(timezone=True), nullable=False),
+    sa.Column('updatedAt', sa.DateTime(timezone=True), nullable=True),
+    sa.PrimaryKeyConstraint('caranteneModifierId')
+    )
+    op.alter_column('item', 'itemId',
+               existing_type=sa.BIGINT(),
+               server_default=sa.Identity(always=True, start=1, increment=1),
+               existing_nullable=False,
+               autoincrement=True)
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.alter_column('item', 'itemId',
+               existing_type=sa.BIGINT(),
+               server_default=sa.Identity(always=False, start=1, increment=1, minvalue=1, maxvalue=9223372036854775807, cycle=False, cache=1),
+               existing_nullable=False,
+               autoincrement=True)
+    op.drop_table('carantene_modifier')
+    # ### end Alembic commands ###

src/backend_api/app/api/api.py

@@ -1,6 +1,8 @@
 from fastapi import APIRouter
 
 from app.api.routes import (
+    carantene_modifier,
+    carantene_modifier_prefix,
     currency,
     currency_prefix,
     item,
@@ -26,6 +28,11 @@
 api_router = APIRouter()
 
 
+api_router.include_router(
+    carantene_modifier.router,
+    prefix=f"/{carantene_modifier_prefix}",
+    tags=[f"{carantene_modifier_prefix}s"],
+)
 api_router.include_router(
     currency.router, prefix=f"/{currency_prefix}", tags=[f"{currency_prefix}s"]
 )
@@ -50,6 +57,9 @@
 api_router.include_router(
     modifier.router, prefix=f"/{modifier_prefix}", tags=[f"{modifier_prefix}s"]
 )
+api_router.include_router(
+    modifier.router, prefix=f"/{modifier_prefix}", tags=[f"{modifier_prefix}s"]
+)
 api_router.include_router(
     plot.router, prefix=f"/{plot_prefix}", tags=[f"{plot_prefix}s"]
 )

src/backend_api/app/api/api_message_util.py

@@ -1,3 +1,5 @@
+from typing import Any
+
 from pydantic import EmailStr
 
 from app.core.schemas.message import Message
@@ -8,7 +10,7 @@
 
 def get_delete_return_msg(
     model_table_name: str,
-    filter: dict[str, str],
+    filter: dict[str, Any] | list[Any],
 ) -> Message:
     """Returns a message indicating the object was deleted successfully.
 
@@ -21,9 +23,11 @@ def get_delete_return_msg(
         str: Message indicating the object was deleted successfully.
     """
 
-    return Message(
-        message=f"{model_table_name} with filter {filter} was deleted successfully"
-    )
+    if isinstance(filter, list):
+        message = f"{model_table_name} with total filter: {len(filter)} and 10 samples: {filter[:10]} was deleted successfully"
+    else:
+        message = f"{model_table_name} with filter {filter} was deleted successfully"
+    return Message(message=message)
 
 
 def get_user_active_change_msg(username: str, active: bool) -> Message:

src/backend_api/app/api/routes/__init__.py

@@ -5,6 +5,7 @@
 from app.api.routes.unidentified_item import unidentified_item_prefix
 from app.api.routes.login import login_prefix
 from app.api.routes.modifier import modifier_prefix
+from app.api.routes.carantene_modifier import carantene_modifier_prefix
 from app.api.routes.plot import plot_prefix
 from app.api.routes.turnstile import turnstile_prefix
 from app.api.routes.user import user_prefix

src/backend_api/app/api/routes/carantene_modifier.py

@@ -0,0 +1,227 @@
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, Query, Request, Response
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+import app.core.schemas as schemas
+from app.api.api_message_util import (
+    get_delete_return_msg,
+)
+from app.api.deps import (
+    get_current_active_superuser,
+    get_current_active_user,
+    get_db,
+)
+from app.api.params import FilterParams
+from app.core.models.models import CaranteneModifier
+from app.core.rate_limit.rate_limit_config import rate_limit_settings
+from app.core.rate_limit.rate_limiters import (
+    apply_user_rate_limits,
+)
+from app.crud import CRUD_carantene_modifier
+
+router = APIRouter()
+
+
+carantene_modifier_prefix = "carantene_modifier"
+
+
+@router.get(
+    "/{caranteneModifierId}",
+    response_model=schemas.CaranteneModifier | list[schemas.CaranteneModifier],
+    dependencies=[Depends(get_current_active_user)],
+)
+@apply_user_rate_limits(
+    rate_limit_settings.DEFAULT_USER_RATE_LIMIT_SECOND,
+    rate_limit_settings.DEFAULT_USER_RATE_LIMIT_MINUTE,
+    rate_limit_settings.DEFAULT_USER_RATE_LIMIT_HOUR,
+    rate_limit_settings.DEFAULT_USER_RATE_LIMIT_DAY,
+)
+async def get_carantene_modifier(
+    request: Request,  # noqa: ARG001
+    response: Response,  # noqa: ARG001
+    caranteneModifierId: int,
+    db: Session = Depends(get_db),
+):
+    """
+    Get carantene modifier or list of carantene modifiers by key and
+    value for "caranteneModifierId"
+
+    Returns one or a list of carantene_modifiers.
+    """
+
+    carantene_modifier_map = {"caranteneModifierId": caranteneModifierId}
+    carantene_modifier = await CRUD_carantene_modifier.get(
+        db=db, filter=carantene_modifier_map
+    )
+
+    return carantene_modifier
+
+
+@router.get(
+    "/",
+    response_model=schemas.CaranteneModifier | list[schemas.CaranteneModifier],
+    dependencies=[Depends(get_current_active_user)],
+)
+@apply_user_rate_limits(
+    rate_limit_settings.DEFAULT_USER_RATE_LIMIT_SECOND,
+    rate_limit_settings.DEFAULT_USER_RATE_LIMIT_MINUTE,
+    rate_limit_settings.DEFAULT_USER_RATE_LIMIT_HOUR,
+    rate_limit_settings.DEFAULT_USER_RATE_LIMIT_DAY,
+)
+async def get_all_carantene_modifiers(
+    request: Request,  # noqa: ARG001
+    response: Response,  # noqa: ARG001
+    filter_params: Annotated[FilterParams, Query()],
+    db: Session = Depends(get_db),
+):
+    """
+    Get all carantene_modifiers.
+
+    Returns a list of all carantene_modifiers.
+    """
+
+    all_carantene_modifiers = await CRUD_carantene_modifier.get(
+        db=db, filter_params=filter_params
+    )
+
+    return all_carantene_modifiers
+
+
+@router.post(
+    "/",
+    response_model=schemas.CaranteneModifierCreate
+    | list[schemas.CaranteneModifierCreate]
+    | None,
+    dependencies=[Depends(get_current_active_superuser)],
+)
+async def create_carantene_modifier(
+    carantene_modifier: schemas.CaranteneModifierCreate
+    | list[schemas.CaranteneModifierCreate],
+    return_nothing: bool | None = None,
+    db: Session = Depends(get_db),
+):
+    """
+    Create one or a list of new carantene_modifiers.
+
+    Returns the created carantene_modifier or list of carantene_modifiers.
+    """
+
+    return await CRUD_carantene_modifier.create(
+        db=db, obj_in=carantene_modifier, return_nothing=return_nothing
+    )
+
+
+@router.put(
+    "/",
+    response_model=schemas.CaranteneModifier,
+    dependencies=[Depends(get_current_active_superuser)],
+)
+async def update_carantene_modifier(
+    caranteneModifierId: int,
+    carantene_modifier_update: schemas.CaranteneModifierUpdate,
+    db: Session = Depends(get_db),
+):
+    """
+    Update a carantene_modifier by key and value for "caranteneModifierId"
+
+    Dominant key is "caranteneModifierId".
+
+    Returns the updated carantene modifier.
+    """
+
+    carantene_modifier_map = {"caranteneModifierId": caranteneModifierId}
+
+    carantene_modifier = await CRUD_carantene_modifier.get(
+        db=db,
+        filter=carantene_modifier_map,
+    )
+
+    return await CRUD_carantene_modifier.update(
+        db_obj=carantene_modifier, obj_in=carantene_modifier_update, db=db
+    )
+
+
+@router.delete(
+    "/{caranteneModifierId}",
+    response_model=str,
+    dependencies=[Depends(get_current_active_superuser)],
+)
+async def delete_carantene_modifier(
+    caranteneModifierId: int,
+    db: Session = Depends(get_db),
+):
+    """
+    Delete a carantene modifier by key and value for "caranteneModifierId"
+
+    Returns a message that the carantene modifier was deleted.
+    Always deletes one carantene_modifier.
+    """
+
+    carantene_modifier_map = {"caranteneModifierId": caranteneModifierId}
+    await CRUD_carantene_modifier.remove(db=db, filter=carantene_modifier_map)
+
+    return get_delete_return_msg(
+        model_table_name=CaranteneModifier.__tablename__, filter=carantene_modifier_map
+    ).message
+
+
+@router.delete(
+    "/bulk-delete/",
+    response_model=str,
+    dependencies=[Depends(get_current_active_superuser)],
+)
+async def bulk_delete_carantene_modifier(
+    caranteneModifierIds: list[schemas.CaranteneModifiersPK],
+    db: Session = Depends(get_db),
+):
+    """
+    Delete a carantene modifier by key and value for "caranteneModifierId"
+
+    Returns a message that the carantene modifier was deleted.
+    Always deletes one carantene_modifier.
+    """
+
+    filter = [car_id.caranteneModifierId for car_id in caranteneModifierIds]
+    await CRUD_carantene_modifier.remove(
+        db=db, filter=filter, deletion_key="caranteneModifierId"
+    )
+
+    return get_delete_return_msg(
+        model_table_name=CaranteneModifier.__tablename__, filter=filter
+    ).message
+
+
+@router.post(
+    "/delete-grouped-dupes/",
+    response_model=str,
+    dependencies=[Depends(get_current_active_superuser)],
+)
+async def delete_grouped_dupes_carantene_modifier(
+    db: Session = Depends(get_db),
+):
+    group_amount = 12
+
+    db.execute(
+        text(
+            f"""
+            DELETE FROM "carantene_modifier"
+            WHERE "caranteneModifierId" IN (
+                SELECT "caranteneModifierId"
+                FROM (
+                    SELECT
+                        "caranteneModifierId",
+                        ROW_NUMBER() OVER (
+                            PARTITION BY "effect"
+                            ORDER BY "createdAt" DESC
+                        ) AS rn
+                    FROM "carantene_modifier"
+                ) AS ranked
+                WHERE rn > {group_amount}
+            )
+            """
+        )
+    )
+
+    return "Successfully deleted duplicate carantene modifier groups by effect"