mirror
/
moonstream
kopia lustrzana https://github.com/bugout-dev/moonstream
1
0
Forkuj 0
Kod Zgłoszenia Projekty Wydania Wiki Aktywność

Merge pull request #891 from moonstream-to/manage-leaderboards 

Manage leaderboards
pull/896/head
Andrey Dolgolev 2023-08-14 18:10:19 +03:00 zatwierdzone przez GitHub
rodzic fbee1a068c 3b52d8f49f
commit f8d067bf7d
Nie znaleziono w bazie danych klucza dla tego podpisu
ID klucza GPG: 4AEE18F83AFDEB23
3 zmienionych plików z 541 dodań i 134 usunięć
Pokaż statystyki Ściągnij plik aktualizacji Ściągnij plik porównania Expand all files Collapse all files

249
engineapi/engineapi/actions.py
Wyświetl plik

@ -1,6 +1,6 @@
from datetime import datetime
from collections import Counter
from typing import List, Any, Optional, Dict, Union
from typing import List, Any, Optional, Dict, Union, Tuple
import uuid
import logging
@ -11,6 +11,7 @@ import requests # type: ignore
from sqlalchemy.dialects.postgresql import insert
from sqlalchemy.orm import Session
from sqlalchemy import func, text, or_
from sqlalchemy.engine import Row
from web3 import Web3
from web3.types import ChecksumAddress
@ -64,6 +65,18 @@ class LeaderboardDeleteScoresError(Exception):
pass
class LeaderboardCreateError(Exception):
pass
class LeaderboardUpdateError(Exception):
pass
class LeaderboardDeleteError(Exception):
pass
BATCH_SIGNATURE_PAGE_SIZE = 500
logger = logging.getLogger(__name__)
@ -208,7 +221,7 @@ def delete_claim(db_session: Session, dropper_claim_id):
"""
claim = (
db_session.query(DropperClaim).filter(DropperClaim.id == dropper_claim_id).one()
db_session.query(DropperClaim).filter(DropperClaim.id == dropper_claim_id).one() # type: ignore
)
db_session.delete(claim)
@ -260,7 +273,7 @@ def activate_drop(db_session: Session, dropper_claim_id: uuid.UUID):
"""
claim = (
db_session.query(DropperClaim).filter(DropperClaim.id == dropper_claim_id).one()
db_session.query(DropperClaim).filter(DropperClaim.id == dropper_claim_id).one() # type: ignore
)
claim.active = True
@ -275,7 +288,7 @@ def deactivate_drop(db_session: Session, dropper_claim_id: uuid.UUID):
"""
claim = (
db_session.query(DropperClaim).filter(DropperClaim.id == dropper_claim_id).one()
db_session.query(DropperClaim).filter(DropperClaim.id == dropper_claim_id).one() # type: ignore
)
claim.active = False
@ -300,7 +313,7 @@ def update_drop(
"""
claim = (
db_session.query(DropperClaim).filter(DropperClaim.id == dropper_claim_id).one()
db_session.query(DropperClaim).filter(DropperClaim.id == dropper_claim_id).one() # type: ignore
)
if title:
@ -619,7 +632,7 @@ def get_drop(db_session: Session, dropper_claim_id: uuid.UUID):
Return particular drop
"""
drop = (
db_session.query(DropperClaim).filter(DropperClaim.id == dropper_claim_id).one()
db_session.query(DropperClaim).filter(DropperClaim.id == dropper_claim_id).one() # type: ignore
)
return drop
@ -833,7 +846,7 @@ def refetch_drop_signatures(
)
.join(DropperContract, DropperClaim.dropper_contract_id == DropperContract.id)
.filter(DropperClaim.id == dropper_claim_id)
).one()
).one() # type: ignore
if claim.claim_block_deadline is None:
raise DropWithNotSettedBlockDeadline(
@ -932,7 +945,7 @@ def refetch_drop_signatures(
return claimant_objects
def get_leaderboard_total_count(db_session: Session, leaderboard_id):
def get_leaderboard_total_count(db_session: Session, leaderboard_id) -> int:
"""
Get the total number of claimants in the leaderboard
"""
@ -943,18 +956,83 @@ def get_leaderboard_total_count(db_session: Session, leaderboard_id):
)
def get_leaderboard(db_session: Session, leaderboard_id: uuid.UUID) -> Leaderboard:
def get_leaderboard_info(
db_session: Session, leaderboard_id: uuid.UUID
) -> Row[Tuple[uuid.UUID, str, str, int, Optional[datetime]]]:
"""
Get the leaderboard from the database
Get the leaderboard from the database with users count
"""
leaderboard = (
db_session.query(Leaderboard).filter(Leaderboard.id == leaderboard_id).one()
db_session.query(
Leaderboard.id,
Leaderboard.title,
Leaderboard.description,
func.count(LeaderboardScores.id).label("users_count"),
func.max(LeaderboardScores.updated_at).label("last_update"),
)
.join(
LeaderboardScores,
LeaderboardScores.leaderboard_id == Leaderboard.id,
isouter=True,
)
.filter(Leaderboard.id == leaderboard_id)
.group_by(Leaderboard.id, Leaderboard.title, Leaderboard.description)
.one()
)
return leaderboard
def get_leaderboard_scores_changes(
db_session: Session, leaderboard_id: uuid.UUID
) -> List[Row[Tuple[int, datetime]]]:
"""
Return the leaderboard scores changes timeline changes of leaderboard scores
"""
leaderboard_scores_changes = (
db_session.query(
func.count(LeaderboardScores.address).label("players_count"),
# func.extract("epoch", LeaderboardScores.updated_at).label("timestamp"),
LeaderboardScores.updated_at.label("date"),
)
.filter(LeaderboardScores.leaderboard_id == leaderboard_id)
.group_by(LeaderboardScores.updated_at)
.order_by(LeaderboardScores.updated_at.desc())
).all()
return leaderboard_scores_changes
def get_leaderboard_scores_by_timestamp(
db_session: Session,
leaderboard_id: uuid.UUID,
date: datetime,
limit: int,
offset: int,
) -> List[LeaderboardScores]:
"""
Return the leaderboard scores by timestamp
"""
leaderboard_scores = (
db_session.query(
LeaderboardScores.leaderboard_id,
LeaderboardScores.address,
LeaderboardScores.score,
LeaderboardScores.points_data,
)
.filter(LeaderboardScores.leaderboard_id == leaderboard_id)
.filter(LeaderboardScores.updated_at == date)
.order_by(LeaderboardScores.score.desc())
.limit(limit)
.offset(offset)
)
return leaderboard_scores
def get_leaderboards(
db_session: Session,
token: Union[str, uuid.UUID],
@ -987,7 +1065,7 @@ def get_leaderboards(
def get_position(
db_session: Session, leaderboard_id, address, window_size, limit: int, offset: int
):
) -> List[Row[Tuple[str, int, int, int, Any]]]:
"""
Return position by address with window size
@ -1039,7 +1117,7 @@ def get_position(
def get_leaderboard_positions(
db_session: Session, leaderboard_id, limit: int, offset: int
):
) -> List[Row[Tuple[uuid.UUID, str, int, str, int]]]:
"""
Get the leaderboard positions
"""
@ -1064,7 +1142,9 @@ def get_leaderboard_positions(
return query
def get_qurtiles(db_session: Session, leaderboard_id):
def get_qurtiles(
db_session: Session, leaderboard_id
) -> Tuple[Row[Tuple[str, float, int]], ...]:
"""
Get the leaderboard qurtiles
https://docs.sqlalchemy.org/en/14/core/functions.html#sqlalchemy.sql.functions.percentile_disc
@ -1098,7 +1178,7 @@ def get_qurtiles(db_session: Session, leaderboard_id):
return q1, q2, q3
def get_ranks(db_session: Session, leaderboard_id):
def get_ranks(db_session: Session, leaderboard_id) -> List[Row[Tuple[int, int, int]]]:
"""
Get the leaderboard rank buckets(rank, size, score)
"""
@ -1126,7 +1206,7 @@ def get_rank(
rank: int,
limit: Optional[int] = None,
offset: Optional[int] = None,
):
) -> List[Row[Tuple[uuid.UUID, str, int, str, int]]]:
"""
Get bucket in leaderboard by rank
"""
@ -1157,33 +1237,114 @@ def get_rank(
return positions
def create_leaderboard(db_session: Session, title: str, description: str):
def create_leaderboard(
db_session: Session,
title: str,
description: Optional[str],
token: Optional[Union[uuid.UUID, str]] = None,
) -> Leaderboard:
"""
Create a leaderboard
"""
leaderboard = Leaderboard(title=title, description=description)
db_session.add(leaderboard)
if not token:
token = uuid.UUID(MOONSTREAM_ADMIN_ACCESS_TOKEN)
try:
leaderboard = Leaderboard(title=title, description=description)
db_session.add(leaderboard)
db_session.commit()
resource = create_leaderboard_resource(
leaderboard_id=str(leaderboard.id),
token=token,
)
leaderboard.resource_id = resource.id
db_session.commit()
except Exception as e:
db_session.rollback()
logger.error(f"Error creating leaderboard: {e}")
raise LeaderboardCreateError(f"Error creating leaderboard: {e}")
return leaderboard
def delete_leaderboard(
db_session: Session, leaderboard_id: uuid.UUID, token: uuid.UUID
) -> Leaderboard:
"""
Delete a leaderboard
"""
try:
leaderboard = (
db_session.query(Leaderboard).filter(Leaderboard.id == leaderboard_id).one() # type: ignore
)
if leaderboard.resource_id is not None:
try:
bc.delete_resource(
token=token,
resource_id=leaderboard.resource_id,
)
except Exception as e:
logger.error(f"Error deleting leaderboard resource: {e}")
else:
logger.error(
f"Leaderboard {leaderboard_id} has no resource id. Skipping. Better delete it manually."
)
db_session.delete(leaderboard)
db_session.commit()
except Exception as e:
db_session.rollback()
logger.error(e)
raise LeaderboardDeleteError(f"Error deleting leaderboard: {e}")
return leaderboard
def update_leaderboard(
db_session: Session,
leaderboard_id: uuid.UUID,
title: Optional[str],
description: Optional[str],
) -> Leaderboard:
"""
Update a leaderboard
"""
leaderboard = (
db_session.query(Leaderboard).filter(Leaderboard.id == leaderboard_id).one() # type: ignore
)
if title is not None:
leaderboard.title = title
if description is not None:
leaderboard.description = description
db_session.commit()
return leaderboard.id
return leaderboard
def get_leaderboard_by_id(db_session: Session, leaderboard_id):
def get_leaderboard_by_id(db_session: Session, leaderboard_id) -> Leaderboard:
"""
Get the leaderboard by id
"""
return db_session.query(Leaderboard).filter(Leaderboard.id == leaderboard_id).one()
return db_session.query(Leaderboard).filter(Leaderboard.id == leaderboard_id).one() # type: ignore
def get_leaderboard_by_title(db_session: Session, title):
def get_leaderboard_by_title(db_session: Session, title) -> Leaderboard:
"""
Get the leaderboard by title
"""
return db_session.query(Leaderboard).filter(Leaderboard.title == title).one()
return db_session.query(Leaderboard).filter(Leaderboard.title == title).one() # type: ignore
def list_leaderboards(db_session: Session, limit: int, offset: int):
def list_leaderboards(
db_session: Session, limit: int, offset: int
) -> List[Row[Tuple[uuid.UUID, str, str]]]:
"""
List all leaderboards
"""
@ -1265,8 +1426,7 @@ def add_scores(
def create_leaderboard_resource(
leaderboard_id: uuid.UUID,
token: Optional[uuid.UUID] = None,
leaderboard_id: str, token: Union[Optional[uuid.UUID], str] = None
) -> BugoutResource:
resource_data: Dict[str, Any] = {
"type": LEADERBOARD_RESOURCE_TYPE,
@ -1275,19 +1435,22 @@ def create_leaderboard_resource(
if token is None:
token = MOONSTREAM_ADMIN_ACCESS_TOKEN
resource = bc.create_resource(
token=MOONSTREAM_ADMIN_ACCESS_TOKEN,
application_id=MOONSTREAM_APPLICATION_ID,
resource_data=resource_data,
timeout=10,
)
try:
resource = bc.create_resource(
token=token,
application_id=MOONSTREAM_APPLICATION_ID,
resource_data=resource_data,
timeout=10,
)
except Exception as e:
raise LeaderboardCreateError(f"Error creating leaderboard resource: {e}")
return resource
def assign_resource(
db_session: Session,
leaderboard_id: uuid.UUID,
user_token: Union[uuid.UUID, str],
resource_id: Optional[uuid.UUID] = None,
):
"""
@ -1295,19 +1458,17 @@ def assign_resource(
"""
leaderboard = (
db_session.query(Leaderboard).filter(Leaderboard.id == leaderboard_id).one()
db_session.query(Leaderboard).filter(Leaderboard.id == leaderboard_id).one() # type: ignore
)
if leaderboard.resource_id is not None:
raise Exception("Leaderboard already has a resource")
if resource_id is not None:
leaderboard.resource_id = resource_id
else:
# Create resource via admin token
resource = create_leaderboard_resource(
leaderboard_id=leaderboard_id,
leaderboard_id=str(leaderboard_id),
token=user_token,
)
leaderboard.resource_id = resource.id
@ -1330,7 +1491,9 @@ def list_leaderboards_resources(
return query.all()
def revoke_resource(db_session: Session, leaderboard_id: uuid.UUID):
def revoke_resource(
db_session: Session, leaderboard_id: uuid.UUID
) -> Optional[uuid.UUID]:
"""
Revoke a resource handler to a leaderboard
"""
@ -1338,7 +1501,7 @@ def revoke_resource(db_session: Session, leaderboard_id: uuid.UUID):
# TODO(ANDREY): Delete resource via admin token
leaderboard = (
db_session.query(Leaderboard).filter(Leaderboard.id == leaderboard_id).one()
db_session.query(Leaderboard).filter(Leaderboard.id == leaderboard_id).one() # type: ignore
)
if leaderboard.resource_id is None:
@ -1354,12 +1517,12 @@ def revoke_resource(db_session: Session, leaderboard_id: uuid.UUID):
def check_leaderboard_resource_permissions(
db_session: Session, leaderboard_id: uuid.UUID, token: uuid.UUID
):
) -> bool:
"""
Check if the user has permissions to access the leaderboard
"""
leaderboard = (
db_session.query(Leaderboard).filter(Leaderboard.id == leaderboard_id).one()
db_session.query(Leaderboard).filter(Leaderboard.id == leaderboard_id).one() # type: ignore
)
permission_url = f"{bc.brood_url}/resources/{leaderboard.resource_id}/holders"

50
engineapi/engineapi/data.py
Wyświetl plik

@ -378,3 +378,53 @@ class LeaderboardInfoResponse(BaseModel):
id: UUID
title: str
description: Optional[str] = None
users_count: int
last_updated_at: Optional[datetime] = None
class LeaderboardCreateRequest(BaseModel):
title: str
description: Optional[str] = None
class LeaderboardCreatedResponse(BaseModel):
id: UUID
title: str
description: Optional[str] = None
resource_id: Optional[UUID] = None
created_at: datetime
updated_at: datetime
class Config:
orm_mode = True
class LeaderboardUpdatedResponse(BaseModel):
id: UUID
title: str
description: Optional[str] = None
resource_id: Optional[UUID] = None
created_at: datetime
updated_at: datetime
class Config:
orm_mode = True
class LeaderboardUpdateRequest(BaseModel):
title: Optional[str] = None
description: Optional[str] = None
class LeaderboardDeletedResponse(BaseModel):
id: UUID
title: str
description: Optional[str] = None
resource_id: Optional[UUID] = None
created_at: datetime
updated_at: datetime
class LeaderboardScoresChangesResponse(BaseModel):
players_count: int
date: datetime

376
engineapi/engineapi/routes/leaderboard.py
Wyświetl plik

@ -1,11 +1,12 @@
"""
Leaderboard API.
"""
from datetime import datetime
import logging
from uuid import UUID
from web3 import Web3
from fastapi import FastAPI, Request, Depends, Response
from fastapi import FastAPI, Request, Depends, Response, Query, Path, Body
from sqlalchemy.orm import Session
from sqlalchemy.orm.exc import NoResultFound
from typing import Any, Dict, List, Optional
@ -33,12 +34,17 @@ leaderboad_whitelist = {
f"/leaderboard/{DOCS_TARGET_PATH}": "GET",
"/leaderboard/openapi.json": "GET",
"/leaderboard/info": "GET",
"/leaderboard/scores/changes": "GET",
"/leaderboard/quartiles": "GET",
"/leaderboard/count/addresses": "GET",
"/leaderboard/position": "GET",
"/leaderboard": "GET",
"/leaderboard/": "GET",
"/leaderboard/rank": "GET",
"/leaderboard/ranks": "GET",
"/scores/changes": "GET",
"/leaderboard/docs": "GET",
"/leaderboard/openapi.json": "GET",
}
app = FastAPI(
@ -62,16 +68,21 @@ app.add_middleware(
)
@app.get("/info", response_model=data.LeaderboardInfoResponse)
async def get_leadeboard(
leaderboard_id: UUID,
@app.get("", response_model=List[data.LeaderboardPosition])
@app.get("/", response_model=List[data.LeaderboardPosition])
async def leaderboard(
leaderboard_id: UUID = Query(..., description="Leaderboard ID"),
limit: int = Query(10),
offset: int = Query(0),
db_session: Session = Depends(db.yield_db_session),
) -> data.LeaderboardInfoResponse:
) -> List[data.LeaderboardPosition]:
"""
Returns leaderboard info.
Returns the leaderboard positions.
"""
### Check if leaderboard exists
try:
leaderboard = actions.get_leaderboard(db_session, leaderboard_id)
actions.get_leaderboard_by_id(db_session, leaderboard_id)
except NoResultFound as e:
raise EngineHTTPException(
status_code=404,
@ -81,10 +92,174 @@ async def get_leadeboard(
logger.error(f"Error while getting leaderboard: {e}")
raise EngineHTTPException(status_code=500, detail="Internal server error")
return data.LeaderboardInfoResponse(
id=leaderboard.id,
title=leaderboard.title,
description=leaderboard.description,
leaderboard_positions = actions.get_leaderboard_positions(
db_session, leaderboard_id, limit, offset
)
result = [
data.LeaderboardPosition(
address=position.address,
score=position.score,
rank=position.rank,
points_data=position.points_data,
)
for position in leaderboard_positions
]
return result
@app.post("", response_model=data.LeaderboardCreatedResponse)
@app.post("/", response_model=data.LeaderboardCreatedResponse)
async def create_leaderboard(
request: Request,
leaderboard: data.LeaderboardCreateRequest = Body(...),
db_session: Session = Depends(db.yield_db_session),
) -> data.LeaderboardCreatedResponse:
"""
Create leaderboard.
"""
token = request.state.token
try:
created_leaderboard = actions.create_leaderboard(
db_session,
title=leaderboard.title,
description=leaderboard.description,
token=token,
)
except actions.LeaderboardCreateError as e:
logger.error(f"Error while creating leaderboard: {e}")
raise EngineHTTPException(
status_code=500,
detail="Leaderboard creation failed. Please try again.",
)
except Exception as e:
logger.error(f"Error while creating leaderboard: {e}")
raise EngineHTTPException(status_code=500, detail="Internal server error")
# Add resource to the leaderboard
return data.LeaderboardCreatedResponse(
id=created_leaderboard.id, # type: ignore
title=created_leaderboard.title, # type: ignore
description=created_leaderboard.description, # type: ignore
resource_id=created_leaderboard.resource_id, # type: ignore
created_at=created_leaderboard.created_at, # type: ignore
updated_at=created_leaderboard.updated_at, # type: ignore
)
@app.put("/{leaderboard_id}", response_model=data.LeaderboardUpdatedResponse)
async def update_leaderboard(
request: Request,
leaderboard_id: UUID = Path(..., description="Leaderboard ID"),
leaderboard: data.LeaderboardUpdateRequest = Body(...),
db_session: Session = Depends(db.yield_db_session),
) -> data.LeaderboardUpdatedResponse:
"""
Update leaderboard.
"""
token = request.state.token
try:
access = actions.check_leaderboard_resource_permissions(
db_session=db_session,
leaderboard_id=leaderboard_id,
token=token,
)
except NoResultFound as e:
raise EngineHTTPException(
status_code=404,
detail="Leaderboard not found.",
)
if access != True:
raise EngineHTTPException(
status_code=403, detail="You don't have access to this leaderboard."
)
try:
updated_leaderboard = actions.update_leaderboard(
db_session=db_session,
leaderboard_id=leaderboard_id,
title=leaderboard.title,
description=leaderboard.description,
)
except actions.LeaderboardUpdateError as e:
logger.error(f"Error while updating leaderboard: {e}")
raise EngineHTTPException(
status_code=500,
detail="Leaderboard update failed. Please try again.",
)
except Exception as e:
logger.error(f"Error while updating leaderboard: {e}")
raise EngineHTTPException(status_code=500, detail="Internal server error")
return data.LeaderboardUpdatedResponse(
id=updated_leaderboard.id, # type: ignore
title=updated_leaderboard.title, # type: ignore
description=updated_leaderboard.description, # type: ignore
resource_id=updated_leaderboard.resource_id, # type: ignore
created_at=updated_leaderboard.created_at, # type: ignore
updated_at=updated_leaderboard.updated_at, # type: ignore
)
@app.delete("/{leaderboard_id}", response_model=data.LeaderboardDeletedResponse)
async def delete_leaderboard(
request: Request,
leaderboard_id: UUID = Path(..., description="Leaderboard ID"),
db_session: Session = Depends(db.yield_db_session),
) -> data.LeaderboardDeletedResponse:
"""
Delete leaderboard.
"""
token = request.state.token
try:
access = actions.check_leaderboard_resource_permissions(
db_session=db_session,
leaderboard_id=leaderboard_id,
token=token,
)
except NoResultFound as e:
raise EngineHTTPException(
status_code=404,
detail="Leaderboard not found.",
)
if access != True:
raise EngineHTTPException(
status_code=403, detail="You don't have access to this leaderboard."
)
try:
deleted_leaderboard = actions.delete_leaderboard(
db_session=db_session,
leaderboard_id=leaderboard_id,
token=token,
)
except actions.LeaderboardDeleteError as e:
logger.error(f"Error while deleting leaderboard: {e}")
raise EngineHTTPException(
status_code=500,
detail="Leaderboard deletion failed. Please try again.",
)
except Exception as e:
logger.error(f"Error while deleting leaderboard: {e}")
raise EngineHTTPException(status_code=500, detail="Internal server error")
return data.LeaderboardDeletedResponse(
id=deleted_leaderboard.id, # type: ignore
title=deleted_leaderboard.title, # type: ignore
description=deleted_leaderboard.description, # type: ignore
created_at=deleted_leaderboard.created_at, # type: ignore
updated_at=deleted_leaderboard.updated_at, # type: ignore
)
@ -111,12 +286,12 @@ async def get_leaderboards(
results = [
data.Leaderboard(
id=leaderboard.id,
title=leaderboard.title,
description=leaderboard.description,
resource_id=leaderboard.resource_id,
created_at=leaderboard.created_at,
updated_at=leaderboard.updated_at,
id=leaderboard.id, # type: ignore
title=leaderboard.title, # type: ignore
description=leaderboard.description, # type: ignore
resource_id=leaderboard.resource_id, # type: ignore
created_at=leaderboard.created_at, # type: ignore
updated_at=leaderboard.updated_at, # type: ignore
)
for leaderboard in leaderboards
]
@ -126,7 +301,7 @@ async def get_leaderboards(
@app.get("/count/addresses", response_model=data.CountAddressesResponse)
async def count_addresses(
leaderboard_id: UUID,
leaderboard_id: UUID = Query(..., description="Leaderboard ID"),
db_session: Session = Depends(db.yield_db_session),
) -> data.CountAddressesResponse:
"""
@ -150,9 +325,64 @@ async def count_addresses(
return data.CountAddressesResponse(count=count)
@app.get("/info", response_model=data.LeaderboardInfoResponse)
async def leadeboard_info(
leaderboard_id: UUID = Query(..., description="Leaderboard ID"),
db_session: Session = Depends(db.yield_db_session),
) -> data.LeaderboardInfoResponse:
"""
Returns leaderboard info.
"""
try:
leaderboard = actions.get_leaderboard_info(db_session, leaderboard_id)
except NoResultFound as e:
raise EngineHTTPException(
status_code=404,
detail="Leaderboard not found.",
)
except Exception as e:
logger.error(f"Error while getting leaderboard: {e}")
raise EngineHTTPException(status_code=500, detail="Internal server error")
return data.LeaderboardInfoResponse(
id=leaderboard.id,
title=leaderboard.title,
description=leaderboard.description,
users_count=leaderboard.users_count,
last_updated_at=leaderboard.last_update,
)
@app.get("/scores/changes")
async def get_scores_changes(
leaderboard_id: UUID = Query(..., description="Leaderboard ID"),
db_session: Session = Depends(db.yield_db_session),
) -> List[data.LeaderboardScoresChangesResponse]:
"""
Returns the score history for the given address.
"""
try:
scores = actions.get_leaderboard_scores_changes(db_session, leaderboard_id)
except actions.LeaderboardIsEmpty:
raise EngineHTTPException(status_code=204, detail="Leaderboard is empty.")
except Exception as e:
logger.error(f"Error while getting scores: {e}")
raise EngineHTTPException(status_code=500, detail="Internal server error")
return [
data.LeaderboardScoresChangesResponse(
players_count=score.players_count,
date=score.date,
)
for score in scores
]
@app.get("/quartiles", response_model=data.QuartilesResponse)
async def quartiles(
leaderboard_id: UUID,
leaderboard_id: UUID = Query(..., description="Leaderboard ID"),
db_session: Session = Depends(db.yield_db_session),
) -> data.QuartilesResponse:
"""
@ -188,12 +418,14 @@ async def quartiles(
@app.get("/position", response_model=List[data.LeaderboardPosition])
async def position(
leaderboard_id: UUID,
address: str,
window_size: int = 1,
limit: int = 10,
offset: int = 0,
normalize_addresses: bool = True,
leaderboard_id: UUID = Query(..., description="Leaderboard ID"),
address: str = Query(..., description="Address to get position for."),
window_size: int = Query(1, description="Amount of positions up and down."),
limit: int = Query(10),
offset: int = Query(0),
normalize_addresses: bool = Query(
True, description="Normalize addresses to checksum."
),
db_session: Session = Depends(db.yield_db_session),
) -> List[data.LeaderboardPosition]:
"""
@ -233,52 +465,12 @@ async def position(
return results
@app.get("", response_model=List[data.LeaderboardPosition])
@app.get("/", response_model=List[data.LeaderboardPosition])
async def leaderboard(
leaderboard_id: UUID,
limit: int = 10,
offset: int = 0,
db_session: Session = Depends(db.yield_db_session),
) -> List[data.LeaderboardPosition]:
"""
Returns the leaderboard positions.
"""
### Check if leaderboard exists
try:
actions.get_leaderboard_by_id(db_session, leaderboard_id)
except NoResultFound as e:
raise EngineHTTPException(
status_code=404,
detail="Leaderboard not found.",
)
except Exception as e:
logger.error(f"Error while getting leaderboard: {e}")
raise EngineHTTPException(status_code=500, detail="Internal server error")
leaderboard_positions = actions.get_leaderboard_positions(
db_session, leaderboard_id, limit, offset
)
result = [
data.LeaderboardPosition(
address=position.address,
score=position.score,
rank=position.rank,
points_data=position.points_data,
)
for position in leaderboard_positions
]
return result
@app.get("/rank", response_model=List[data.LeaderboardPosition])
async def rank(
leaderboard_id: UUID,
rank: int = 1,
limit: Optional[int] = None,
offset: Optional[int] = None,
leaderboard_id: UUID = Query(..., description="Leaderboard ID"),
rank: int = Query(1, description="Rank to get."),
limit: Optional[int] = Query(None),
offset: Optional[int] = Query(None),
db_session: Session = Depends(db.yield_db_session),
) -> List[data.LeaderboardPosition]:
"""
@ -314,7 +506,8 @@ async def rank(
@app.get("/ranks", response_model=List[data.RanksResponse])
async def ranks(
leaderboard_id: UUID, db_session: Session = Depends(db.yield_db_session)
leaderboard_id: UUID = Query(..., description="Leaderboard ID"),
db_session: Session = Depends(db.yield_db_session),
) -> List[data.RanksResponse]:
"""
Returns the leaderboard rank buckets overview with score and size of bucket.
@ -347,38 +540,39 @@ async def ranks(
@app.put("/{leaderboard_id}/scores", response_model=List[data.LeaderboardScore])
async def leaderboard_push_scores(
request: Request,
leaderboard_id: UUID,
scores: List[data.Score],
overwrite: bool = False,
normalize_addresses: bool = True,
leaderboard_id: UUID = Path(..., description="Leaderboard ID"),
scores: List[data.Score] = Body(
..., description="Scores to put to the leaderboard."
),
overwrite: bool = Query(
False,
description="If enabled, this will delete all current scores and replace them with the new scores provided.",
),
normalize_addresses: bool = Query(
True, description="Normalize addresses to checksum."
),
db_session: Session = Depends(db.yield_db_session),
) -> List[data.LeaderboardScore]:
"""
Put the leaderboard to the database.
"""
access = actions.check_leaderboard_resource_permissions(
db_session=db_session,
leaderboard_id=leaderboard_id,
token=request.state.token,
)
if not access:
raise EngineHTTPException(
status_code=403, detail="You don't have access to this leaderboard."
)
### Check if leaderboard exists
token = request.state.token
try:
actions.get_leaderboard_by_id(db_session, leaderboard_id)
access = actions.check_leaderboard_resource_permissions(
db_session=db_session,
leaderboard_id=leaderboard_id,
token=token,
)
except NoResultFound as e:
raise EngineHTTPException(
status_code=404,
detail="Leaderboard not found.",
)
except Exception as e:
logger.error(f"Error while getting leaderboard: {e}")
raise EngineHTTPException(status_code=500, detail="Internal server error")
if not access:
raise EngineHTTPException(
status_code=403, detail="You don't have access to this leaderboard."
)
try:
leaderboard_points = actions.add_scores(