Add ltd2 client (#1)

Add ltd2 client and model, loading data from ltd2 api ready

Author: Pompeiro

.env.template

@@ -36,3 +36,6 @@ SENTRY_DSN=
 # Configure these with your own Docker registry images
 DOCKER_IMAGE_BACKEND=backend
 DOCKER_IMAGE_FRONTEND=frontend
+
+LTD2_X_API_KEY=changethis
+LTD2_ICONS_PATH=changethis

README.md

@@ -1,3 +1,41 @@
+# LTD2-DSS
+copy env file
+```bash
+cp .env.template .env
+```
+
+Fill env variable with your API key from https://developer.legiontd2.com/home - key_management
+```
+LTD2_X_API_KEY=xxx
+```
+Fill env variable with your path to game icons, use following example:
+```
+LTD2_ICONS_PATH="/home/maciej/.steam/debian-installation/steamapps/common/Legion TD 2/Legion TD 2_Data/uiresources/AeonGT/hud/img/icons/"
+```
+
+feed environment variables from .env
+```bash
+source .env
+```
+change owner of files
+```bash
+sudo chown -R $(whoami) .
+```
+
+open psql where a84 is postgres container id
+```bash
+docker exec -it a84 psql -U postgres -d app
+```
+clear docker
+```bash
+docker system prune -a --volumes
+```
+
+run project
+```bash
+docker compose up -d
+```
+
 # Full Stack FastAPI Template
 
 <a href="https://github.com/tiangolo/full-stack-fastapi-template/actions?query=workflow%3ATest" target="_blank"><img src="https://github.com/tiangolo/full-stack-fastapi-template/workflows/Test/badge.svg" alt="Test"></a>

backend/app/api/deps.py

@@ -4,9 +4,11 @@
 from fastapi import Depends, HTTPException, status
 from fastapi.security import OAuth2PasswordBearer
 from jose import JWTError, jwt
+from playwright.async_api import APIRequestContext
 from pydantic import ValidationError
 from sqlmodel import Session
 
+from app.clients.ltd2 import get_api_request_context
 from app.core import security
 from app.core.config import settings
 from app.core.db import engine
@@ -23,6 +25,7 @@ def get_db() -> Generator[Session, None, None]:
 
 
 SessionDep = Annotated[Session, Depends(get_db)]
+APIRequestContextDep = Annotated[APIRequestContext, Depends(get_api_request_context)]
 TokenDep = Annotated[str, Depends(reusable_oauth2)]
 
 
@@ -51,6 +54,6 @@ def get_current_user(session: SessionDep, token: TokenDep) -> User:
 def get_current_active_superuser(current_user: CurrentUser) -> User:
     if not current_user.is_superuser:
         raise HTTPException(
-            status_code=400, detail="The user doesn't have enough privileges"
+            status_code=403, detail="The user doesn't have enough privileges"
         )
     return current_user

backend/app/api/main.py

@@ -1,9 +1,10 @@
 from fastapi import APIRouter
 
-from app.api.routes import items, login, users, utils
+from app.api.routes import items, login, ltd2_units, users, utils
 
 api_router = APIRouter()
 api_router.include_router(login.router, tags=["login"])
 api_router.include_router(users.router, prefix="/users", tags=["users"])
 api_router.include_router(utils.router, prefix="/utils", tags=["utils"])
 api_router.include_router(items.router, prefix="/items", tags=["items"])
+api_router.include_router(ltd2_units.router, prefix="/ltd2-units", tags=["ltd2-units"])

backend/app/api/routes/items.py

@@ -4,12 +4,12 @@
 from sqlmodel import func, select
 
 from app.api.deps import CurrentUser, SessionDep
-from app.models import Item, ItemCreate, ItemOut, ItemsOut, ItemUpdate, Message
+from app.models import Item, ItemCreate, ItemPublic, ItemsPublic, ItemUpdate, Message
 
 router = APIRouter()
 
 
-@router.get("/", response_model=ItemsOut)
+@router.get("/", response_model=ItemsPublic)
 def read_items(
     session: SessionDep, current_user: CurrentUser, skip: int = 0, limit: int = 100
 ) -> Any:
@@ -37,10 +37,10 @@ def read_items(
         )
         items = session.exec(statement).all()
 
-    return ItemsOut(data=items, count=count)
+    return ItemsPublic(data=items, count=count)
 
 
-@router.get("/{id}", response_model=ItemOut)
+@router.get("/{id}", response_model=ItemPublic)
 def read_item(session: SessionDep, current_user: CurrentUser, id: int) -> Any:
     """
     Get item by ID.
@@ -53,7 +53,7 @@ def read_item(session: SessionDep, current_user: CurrentUser, id: int) -> Any:
     return item
 
 
-@router.post("/", response_model=ItemOut)
+@router.post("/", response_model=ItemPublic)
 def create_item(
     *, session: SessionDep, current_user: CurrentUser, item_in: ItemCreate
 ) -> Any:
@@ -67,7 +67,7 @@ def create_item(
     return item
 
 
-@router.put("/{id}", response_model=ItemOut)
+@router.put("/{id}", response_model=ItemPublic)
 def update_item(
     *, session: SessionDep, current_user: CurrentUser, id: int, item_in: ItemUpdate
 ) -> Any:

backend/app/api/routes/login.py

@@ -10,7 +10,7 @@
 from app.core import security
 from app.core.config import settings
 from app.core.security import get_password_hash
-from app.models import Message, NewPassword, Token, UserOut
+from app.models import Message, NewPassword, Token, UserPublic
 from app.utils import (
     generate_password_reset_token,
     generate_reset_password_email,
@@ -43,7 +43,7 @@ def login_access_token(
     )
 
 
-@router.post("/login/test-token", response_model=UserOut)
+@router.post("/login/test-token", response_model=UserPublic)
 def test_token(current_user: CurrentUser) -> Any:
     """
     Test access token

backend/app/api/routes/ltd2_units.py

@@ -0,0 +1,38 @@
+from collections.abc import Sequence
+from http import HTTPStatus
+
+from fastapi import APIRouter, Query
+from sqlmodel import select
+
+from app.api.deps import APIRequestContextDep, SessionDep
+from app.clients.ltd2 import (
+    find_units_by_version,
+    get_current_game_version,
+)
+from app.models import LTD2Unit
+
+router = APIRouter()
+
+
+@router.post("/", status_code=HTTPStatus.CREATED)
+async def clone_units_from_ltd2_api(
+    session: SessionDep,
+    api_request_context: APIRequestContextDep,
+) -> None:
+    version = await get_current_game_version(api_request_context=api_request_context)
+    units_list = await find_units_by_version(
+        version=version, api_request_context=api_request_context
+    )
+    ltd2_units = [LTD2Unit.model_validate(unit) for unit in units_list]
+
+    session.bulk_save_objects(ltd2_units)
+    session.commit()
+    session.expire_all()
+
+
+@router.get("/")
+async def read_units(
+    session: SessionDep, offset: int = 0, limit: int = Query(default=10, le=500)
+) -> Sequence[LTD2Unit]:
+    units = session.exec(select(LTD2Unit).offset(offset).limit(limit)).all()
+    return units

backend/app/api/routes/users.py

@@ -17,9 +17,9 @@
     UpdatePassword,
     User,
     UserCreate,
-    UserOut,
+    UserPublic,
     UserRegister,
-    UsersOut,
+    UsersPublic,
     UserUpdate,
     UserUpdateMe,
 )
@@ -29,7 +29,9 @@
 
 
 @router.get(
-    "/", dependencies=[Depends(get_current_active_superuser)], response_model=UsersOut
+    "/",
+    dependencies=[Depends(get_current_active_superuser)],
+    response_model=UsersPublic,
 )
 def read_users(session: SessionDep, skip: int = 0, limit: int = 100) -> Any:
     """
@@ -42,11 +44,11 @@ def read_users(session: SessionDep, skip: int = 0, limit: int = 100) -> Any:
     statement = select(User).offset(skip).limit(limit)
     users = session.exec(statement).all()
 
-    return UsersOut(data=users, count=count)
+    return UsersPublic(data=users, count=count)
 
 
 @router.post(
-    "/", dependencies=[Depends(get_current_active_superuser)], response_model=UserOut
+    "/", dependencies=[Depends(get_current_active_superuser)], response_model=UserPublic
 )
 def create_user(*, session: SessionDep, user_in: UserCreate) -> Any:
     """
@@ -72,7 +74,7 @@ def create_user(*, session: SessionDep, user_in: UserCreate) -> Any:
     return user
 
 
-@router.patch("/me", response_model=UserOut)
+@router.patch("/me", response_model=UserPublic)
 def update_user_me(
     *, session: SessionDep, user_in: UserUpdateMe, current_user: CurrentUser
 ) -> Any:
@@ -114,15 +116,31 @@ def update_password_me(
     return Message(message="Password updated successfully")
 
 
-@router.get("/me", response_model=UserOut)
+@router.get("/me", response_model=UserPublic)
 def read_user_me(current_user: CurrentUser) -> Any:
     """
     Get current user.
     """
     return current_user
 
 
-@router.post("/signup", response_model=UserOut)
+@router.delete("/me", response_model=Message)
+def delete_user_me(session: SessionDep, current_user: CurrentUser) -> Any:
+    """
+    Delete own user.
+    """
+    if current_user.is_superuser:
+        raise HTTPException(
+            status_code=403, detail="Super users are not allowed to delete themselves"
+        )
+    statement = delete(Item).where(col(Item.owner_id) == current_user.id)
+    session.exec(statement)  # type: ignore
+    session.delete(current_user)
+    session.commit()
+    return Message(message="User deleted successfully")
+
+
+@router.post("/signup", response_model=UserPublic)
 def register_user(session: SessionDep, user_in: UserRegister) -> Any:
     """
     Create new user without the need to be logged in.
@@ -143,7 +161,7 @@ def register_user(session: SessionDep, user_in: UserRegister) -> Any:
     return user
 
 
-@router.get("/{user_id}", response_model=UserOut)
+@router.get("/{user_id}", response_model=UserPublic)
 def read_user_by_id(
     user_id: int, session: SessionDep, current_user: CurrentUser
 ) -> Any:
@@ -164,7 +182,7 @@ def read_user_by_id(
 @router.patch(
     "/{user_id}",
     dependencies=[Depends(get_current_active_superuser)],
-    response_model=UserOut,
+    response_model=UserPublic,
 )
 def update_user(
     *,
@@ -193,7 +211,7 @@ def update_user(
     return db_user
 
 
-@router.delete("/{user_id}")
+@router.delete("/{user_id}", dependencies=[Depends(get_current_active_superuser)])
 def delete_user(
     session: SessionDep, current_user: CurrentUser, user_id: int
 ) -> Message:
@@ -203,15 +221,10 @@ def delete_user(
     user = session.get(User, user_id)
     if not user:
         raise HTTPException(status_code=404, detail="User not found")
-    elif user != current_user and not current_user.is_superuser:
-        raise HTTPException(
-            status_code=403, detail="The user doesn't have enough privileges"
-        )
-    elif user == current_user and current_user.is_superuser:
+    if user == current_user:
         raise HTTPException(
             status_code=403, detail="Super users are not allowed to delete themselves"
         )
-
     statement = delete(Item).where(col(Item.owner_id) == user_id)
     session.exec(statement)  # type: ignore
     session.delete(user)

backend/app/clients/__init__.py

@@ -0,0 +1,53 @@
+from collections.abc import AsyncGenerator
+
+from playwright.async_api import APIRequestContext, async_playwright
+
+from app.core.config import settings
+from app.models import LTD2UnitFromAPI
+
+assert settings.LTD2_X_API_KEY, "LTD2 X API KEY not set"
+
+
+async def get_api_request_context() -> AsyncGenerator[APIRequestContext, None]:
+    async with async_playwright() as playwright:
+        headers = {
+            "x-api-key": settings.LTD2_X_API_KEY,
+        }
+        request_context = await playwright.request.new_context(
+            base_url="https://apiv2.legiontd2.com", extra_http_headers=headers
+        )
+        yield request_context
+
+
+async def find_unit_by_name(name: str, api_request_context: APIRequestContext) -> dict:  # type:ignore
+    unit = await api_request_context.get(f"/units/byName/{name}")
+    unit_response = await unit.json()
+    return unit_response  # type:ignore
+
+
+async def get_current_game_version(api_request_context: APIRequestContext) -> str:
+    unit_response = await find_unit_by_name(
+        name="Berserker", api_request_context=api_request_context
+    )
+    version = unit_response["version"]
+    return version  # type:ignore
+
+
+async def find_units_by_version(
+    version: str, api_request_context: APIRequestContext
+) -> list[LTD2UnitFromAPI]:
+    units = await api_request_context.get(
+        f"/units/byVersion/{version}", params={"limit": 250}
+    )
+    units_response = await units.json()
+    if len(units_response) == 250:
+        units_remaining = await api_request_context.get(
+            f"/units/byVersion/{version}", params={"limit": 250, "offset": 250}
+        )
+        units_remaining_response = await units_remaining.json()
+        units_response = units_response + units_remaining_response
+
+    return [
+        LTD2UnitFromAPI.model_validate(LTD2UnitFromAPI(**unit))
+        for unit in units_response
+    ]