Add get score endpoint.

pull/995/head
Andrey 2024-01-10 17:25:30 +02:00
rodzic 854a473043
commit cfeae159d4
2 zmienionych plików z 89 dodań i 0 usunięć

Wyświetl plik

@ -1222,6 +1222,43 @@ def get_position(
return query.all()
def get_leaderboard_score(
db_session: Session,
leaderboard_id,
address,
version_number: Optional[int] = None,
) -> Optional[LeaderboardScores]:
"""
Return address score
"""
latest_version = leaderboard_version_filter(
db_session=db_session,
leaderboard_id=leaderboard_id,
version_number=version_number,
)
query = (
db_session.query(LeaderboardScores)
.join(
LeaderboardVersion,
and_(
LeaderboardVersion.leaderboard_id == LeaderboardScores.leaderboard_id,
LeaderboardVersion.version_number
== LeaderboardScores.leaderboard_version_number,
),
)
.filter(
LeaderboardVersion.published == True,
LeaderboardVersion.version_number == latest_version,
)
.filter(LeaderboardScores.leaderboard_id == leaderboard_id)
.filter(LeaderboardScores.address == address)
)
return query.one_or_none()
def get_leaderboard_positions(
db_session: Session,
leaderboard_id,

Wyświetl plik

@ -60,6 +60,7 @@ leaderboard_whitelist = {
"/leaderboard/": "GET",
"/leaderboard/rank": "GET",
"/leaderboard/ranks": "GET",
"/leaderboard/scores": "GET",
"/scores/changes": "GET",
"/leaderboard/docs": "GET",
"/leaderboard/openapi.json": "GET",
@ -619,6 +620,57 @@ async def ranks(
return results
@app.get(
"/scores",
response_model=data.LeaderboardScore,
tags=["Public Endpoints"],
)
async def leaderboard_score(
address: str = Query(..., description="Address to get position for."),
leaderboard_id: UUID = Query(..., description="Leaderboard ID"),
version: Optional[int] = Query(None, description="Version of the leaderboard."),
normalize_addresses: bool = Query(
True, description="Normalize addresses to checksum."
),
db_session: Session = Depends(db.yield_db_session),
) -> data.LeaderboardScore:
"""
Returns the leaderboard posotion for the given address.
"""
### 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")
if normalize_addresses:
address = Web3.toChecksumAddress(address)
score = actions.get_leaderboard_score(
db_session,
leaderboard_id,
address,
version,
)
if score is None:
raise EngineHTTPException(status_code=204, detail="Score not found.")
return data.LeaderboardScore(
leaderboard_id=score.leaderboard_id,
address=score.address,
score=score.score,
points_data=score.points_data,
)
@app.put(
"/{leaderboard_id}/scores",
response_model=List[data.LeaderboardScore],