Merge pull request #509 from Bot-detector/develop

Release

docker-compose.yml

@@ -17,18 +17,18 @@ services:
     networks:
       - botdetector-network
 
-    kafka:
-        image: bitnami/kafka:latest
-        environment:
-          - ALLOW_PLAINTEXT_LISTENER=yes
-          - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://:9094
-          - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,EXTERNAL:PLAINTEXT
-          - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092,EXTERNAL://localhost:9094
-          - KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=true
-        ports:
-          - 9094:9094
-        networks:
-          - botdetector-network
+  kafka:
+      image: bitnami/kafka:latest
+      environment:
+        - ALLOW_PLAINTEXT_LISTENER=yes
+        - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093,EXTERNAL://:9094
+        - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,EXTERNAL:PLAINTEXT
+        - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092,EXTERNAL://localhost:9094
+        - KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=true
+      ports:
+        - 9094:9094
+      networks:
+        - botdetector-network
 
   api:
     build:

src/api/v1/hiscore.py

@@ -193,6 +193,7 @@ async def get_latest_hiscore_data_by_player_features(
     confirmed_player: Optional[int] = Query(None, ge=0, le=1),
     label_id: Optional[int] = Query(None, ge=0),
     label_jagex: Optional[int] = Query(None, ge=0, le=5),
+    greater_than: Optional[int] = Query(None, ge=0),
 ):
     """
     Select the latest hiscore data of multiple players by filtering on the player features.
@@ -211,6 +212,7 @@ async def get_latest_hiscore_data_by_player_features(
         == confirmed_player
         == label_id
         == label_jagex
+        == greater_than
     ):
         raise HTTPException(status_code=404, detail="No param given")
 
@@ -233,6 +235,9 @@ async def get_latest_hiscore_data_by_player_features(
     if not label_jagex is None:
         sql = sql.where(Player.label_jagex == label_jagex)
 
+    if not greater_than is None:
+        sql = sql.where(Player.id >= greater_than)
+
     # paging
     sql = sql.limit(row_count).offset(row_count * (page - 1))
 

src/api/v2/highscore.py

@@ -4,6 +4,9 @@
 from src.app.repositories.highscore import (
     PlayerHiscoreData as RepositoryPlayerHiscoreData,
 )
+from src.app.repositories.highscore_latest import (
+    PlayerHiscoreDataLatest as RepositoryPlayerHiscoreDataLatest,
+)
 from src.app.schemas.highscore import PlayerHiscoreData as SchemaPlayerHiscoreData
 from pydantic import BaseModel
 
@@ -29,6 +32,25 @@ async def get_highscore_data(
     return data
 
 
+@router.get("/hiscore/latest", response_model=list[SchemaPlayerHiscoreData])
+async def get_highscore_data_latest(
+    request: Request,
+    gte_player_id: int = Query(ge=0, description="player id greater than or equal to"),
+    page: int = Query(default=None, ge=1),
+    page_size: int = Query(default=1000, ge=1),
+    token: str = Header(...),
+):
+    await verify_token(
+        token,
+        verification="verify_ban",
+        route=logging_helpers.build_route_log_string(request),
+    )
+
+    repo = RepositoryPlayerHiscoreDataLatest()
+    data = await repo.read(gte_player_id=gte_player_id, page=page, page_size=page_size)
+    return data
+
+
 @router.post("/hiscore", status_code=status.HTTP_201_CREATED)
 async def post_highscore_data(
     request: Request, data: list[SchemaPlayerHiscoreData], token: str = Header(...)

src/api/v2/player.py

@@ -1,9 +1,10 @@
-from fastapi import APIRouter, Query, status, Header, Request
-from src.database.functions import verify_token
-from src.utils import logging_helpers
+from fastapi import APIRouter, Header, Query, Request, status
+from fastapi.exceptions import HTTPException
+
 from src.app.repositories.player import Player as RepositoryPlayer
 from src.app.schemas.player import Player as SchemaPlayer
-from pydantic import BaseModel
+from src.database.functions import verify_token
+from src.utils import logging_helpers
 
 router = APIRouter(tags=["Player"])
 
@@ -30,8 +31,9 @@ async def get_player_data(
 @router.get("/players", response_model=list[SchemaPlayer])
 async def get_many_players_data(
     request: Request,
-    page: int = Query(default=1, ge=1),
+    page: int = Query(default=None),
     page_size: int = Query(default=10, ge=1, le=10_000),
+    greater_than: int = Query(default=None),
     token: str = Header(...),
 ):
     await verify_token(
@@ -40,8 +42,22 @@ async def get_many_players_data(
         route=logging_helpers.build_route_log_string(request),
     )
 
+    if greater_than is None and page is None:
+        raise HTTPException(
+            status_code=status.HTTP_406_NOT_ACCEPTABLE,
+            detail=f"page or greater than required, received: {page=}, {greater_than=}",
+        )
+
+    if all([page, greater_than]):
+        raise HTTPException(
+            status_code=status.HTTP_406_NOT_ACCEPTABLE,
+            detail=f"either page or greater than not both can be set, received: {page=}, {greater_than=}",
+        )
+
     repo = RepositoryPlayer()
-    data = await repo.read_many(page=page, page_size=page_size)
+    data = await repo.read_many(
+        page=page, page_size=page_size, greater_than=greater_than
+    )
     return data
 
 

src/app/repositories/highscore.py

@@ -1,19 +1,17 @@
 import logging
 
-from pydantic import ValidationError
-from sqlalchemy import delete, select, update, union_all
+from pydantic import BaseModel, ValidationError
+from sqlalchemy import select, union_all
 from sqlalchemy.dialects.mysql import insert
 from sqlalchemy.exc import OperationalError
 from sqlalchemy.ext.asyncio import AsyncResult, AsyncSession
-from sqlalchemy.sql.expression import Delete, Insert, Select, Update, and_
+from sqlalchemy.sql.expression import Insert, Select, and_
 
 from src.app.schemas.highscore import PlayerHiscoreData as SchemaHiscore
-from src.app.schemas.player import Player as SchemaPlayer
 from src.database.database import PLAYERDATA_ENGINE
+from src.database.functions import handle_database_error
 from src.database.models import Player as dbPlayer
 from src.database.models import playerHiscoreData as dbPlayerHiscoreData
-from src.database.functions import handle_database_error
-from pydantic import BaseModel
 
 logger = logging.getLogger(__name__)
 
@@ -43,7 +41,7 @@ async def _get_unique(self, data: list[SchemaHiscore]) -> list[SchemaHiscore]:
         # Get an async session from the PLAYERDATA_ENGINE.
         async with PLAYERDATA_ENGINE.get_session() as session:
             # Cast the session to AsyncSession for type hinting.
-            session: AsyncSession = session
+            session: AsyncSession
 
             # Execute the final_query and fetch the results.
             result: AsyncResult = await session.execute(final_query)

src/app/repositories/highscore_latest.py

@@ -0,0 +1,57 @@
+import logging
+
+from fastapi.encoders import jsonable_encoder
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.app.schemas.highscore import PlayerHiscoreData as SchemaHiscore
+from src.database.database import PLAYERDATA_ENGINE
+from src.database.functions import handle_database_error
+from src.database.models import Player as dbPlayer
+from src.database.models import PlayerHiscoreDataLatest as dbPlayerHiscoreDataLatest
+
+logger = logging.getLogger(__name__)
+
+
+class PlayerHiscoreDataLatest:
+    def __init__(self):
+        pass
+
+    @handle_database_error
+    async def read(
+        self, gte_player_id: int = None, page: int = None, page_size: int = None
+    ):
+        table = dbPlayerHiscoreDataLatest
+
+        sql_select = select(table)
+        sql_select = sql_select.join(
+            target=dbPlayer, onclause=table.Player_id == dbPlayer.id
+        )
+        sql_select = sql_select.order_by(table.Player_id.asc())
+
+        if gte_player_id:
+            sql_select = sql_select.where(table.Player_id >= gte_player_id)
+
+        if page_size:
+            sql_select = sql_select.limit(page_size)
+
+        if page is not None and page_size:
+            sql_select = sql_select.offset((page - 1) * page_size)
+
+        async with PLAYERDATA_ENGINE.get_session() as session:
+            session: AsyncSession
+            # Execute the select query
+            result = await session.execute(sql_select)
+
+        # Convert the query results to SchemaHiscore objects
+        schema_data = []
+        for row in result.scalars().all():
+            try:
+                row_dict: dict = jsonable_encoder(row)
+                row_dict = {k: v if v is not None else 0 for k, v in row_dict.items()}
+                model_row = SchemaHiscore.model_validate(row_dict)
+                schema_data.append(model_row)
+            except Exception as e:
+                logger.error({"error": str(e), "data": row_dict})
+
+        return schema_data

src/app/repositories/player.py

@@ -112,14 +112,21 @@ async def delete(self, player_name: str):
         pass
 
     @handle_database_error
-    async def read_many(self, page: int = 1, page_size: int = 10):
+    async def read_many(
+        self, page: int = None, page_size: int = 10_000, greater_than: int = None
+    ):
         table = dbPlayer
 
         sql_select: Select = select(table)
-        sql_select = sql_select.limit(page_size).offset((page - 1) * page_size)
+
+        if greater_than:
+            sql_select = sql_select.where(table.id > greater_than).limit(page_size)
+
+        if page:
+            sql_select = sql_select.limit(page_size).offset((page - 1) * page_size)
 
         async with PLAYERDATA_ENGINE.get_session() as session:
-            session: AsyncSession = session
+            session: AsyncSession
             # Execute the select query
             result: AsyncResult = await session.execute(sql_select)
 
@@ -129,5 +136,5 @@ async def read_many(self, page: int = 1, page_size: int = 10):
             try:
                 schema_data.append(SchemaPlayer.model_validate(row))
             except ValidationError as e:
-                print(e)
+                logger.error(e)
         return schema_data

src/app/schemas/highscore.py

@@ -81,7 +81,7 @@ class PlayerHiscoreData(BaseModel):
     sarachnis: int
     scorpia: int
     skotizo: int
-    tempoross: int
+    tempoross: int = 0
     the_gauntlet: int
     the_corrupted_gauntlet: int
     theatre_of_blood: int

src/core/config.py

@@ -13,7 +13,8 @@
 sql_uri = os.environ.get("sql_uri")
 discord_sql_uri = os.environ.get("discord_sql_uri")
 token = os.environ.get("token")
-kafka_url = os.environ.get("kafka_url", "localhost:9094")
+kafka_url = os.environ.get("kafka_url", "127.0.0.1:9094")
+env = os.environ.get("env", "DEV")
 
 # setup logging
 file_handler = logging.FileHandler(filename="./error.log", mode="a")

src/core/server.py

@@ -98,8 +98,9 @@ async def validation_exception_handler(request: Request, exc: RequestValidationE
     return JSONResponse(content={"detail": error}, status_code=422)
 
 
-@app.on_event("startup")
-async def startup_event():
-    logger.info("startup initiated")
-    highscore_processor = HighscoreProcessor(batch_size=100)
-    asyncio.ensure_future(highscore_processor.start())
+# @app.on_event("startup")
+# async def startup_event():
+#     if config.env != "DEV":
+#         logger.info("startup initiated")
+#         highscore_processor = HighscoreProcessor(batch_size=100)
+#         asyncio.ensure_future(highscore_processor.start())