kopia lustrzana https://github.com/bugout-dev/moonstream
rodzic
73bc3bdc70
commit
d27e675df0
|
@ -1,6 +1,6 @@
|
|||
from datetime import datetime
|
||||
from collections import Counter
|
||||
from typing import List, Any, Optional, Dict
|
||||
from typing import List, Any, Optional, Dict, Union
|
||||
import uuid
|
||||
import logging
|
||||
|
||||
|
@ -14,7 +14,7 @@ from sqlalchemy import func, text, or_
|
|||
from web3 import Web3
|
||||
from web3.types import ChecksumAddress
|
||||
|
||||
from .data import Score
|
||||
from .data import Score, LeaderboardScore
|
||||
from .contracts import Dropper_interface, ERC20_interface, Terminus_interface
|
||||
from .models import (
|
||||
DropperClaimant,
|
||||
|
@ -33,6 +33,10 @@ from .settings import (
|
|||
)
|
||||
|
||||
|
||||
class LeaderboardsResourcesNotFound(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class AuthorizationError(Exception):
|
||||
pass
|
||||
|
||||
|
@ -939,6 +943,48 @@ def get_leaderboard_total_count(db_session: Session, leaderboard_id):
|
|||
)
|
||||
|
||||
|
||||
def get_leaderboard(db_session: Session, leaderboard_id: uuid.UUID):
|
||||
"""
|
||||
Get the leaderboard from the database
|
||||
"""
|
||||
|
||||
leaderboard = (
|
||||
db_session.query(Leaderboard).filter(Leaderboard.id == leaderboard_id).one()
|
||||
)
|
||||
|
||||
return leaderboard
|
||||
|
||||
|
||||
def get_leaderboards(
|
||||
db_session: Session,
|
||||
token: Union[str, uuid.UUID],
|
||||
):
|
||||
"""
|
||||
Get the leaderboards resources
|
||||
"""
|
||||
|
||||
user_resources = bc.list_resources(
|
||||
token=token,
|
||||
params={"type": "leaderboard"},
|
||||
)
|
||||
|
||||
if len(user_resources.resources) == 0:
|
||||
raise LeaderboardsResourcesNotFound(f"Leaderboard not found for token")
|
||||
|
||||
leaderboards_ids = []
|
||||
|
||||
for resource in user_resources.resources:
|
||||
leaderboard_id = resource.resource_data["leaderboard_id"]
|
||||
|
||||
leaderboards_ids.append(leaderboard_id)
|
||||
|
||||
leaderboards = (
|
||||
db_session.query(Leaderboard).filter(Leaderboard.id.in_(leaderboards_ids)).all()
|
||||
)
|
||||
|
||||
return leaderboards
|
||||
|
||||
|
||||
def get_position(
|
||||
db_session: Session, leaderboard_id, address, window_size, limit: int, offset: int
|
||||
):
|
||||
|
|
|
@ -320,3 +320,25 @@ class RanksResponse(BaseModel):
|
|||
rank: int
|
||||
score: int
|
||||
size: int
|
||||
|
||||
|
||||
class LeaderboardScore(BaseModel):
|
||||
leaderboard_id: UUID
|
||||
address: str
|
||||
score: int
|
||||
points_data: Dict[str, Any]
|
||||
|
||||
|
||||
class Leaderboard(BaseModel):
|
||||
id: UUID
|
||||
title: str
|
||||
description: Optional[str] = None
|
||||
resource_id: Optional[UUID] = None
|
||||
created_at: datetime
|
||||
updated_at: datetime
|
||||
|
||||
|
||||
class LeaderboardInfoResponse(BaseModel):
|
||||
id: UUID
|
||||
title: str
|
||||
description: Optional[str] = None
|
||||
|
|
|
@ -30,6 +30,7 @@ tags_metadata = [
|
|||
|
||||
|
||||
leaderboad_whitelist = {
|
||||
"/leaderboard/info": "GET",
|
||||
"/leaderboard/quartiles": "GET",
|
||||
"/leaderboard/count/addresses": "GET",
|
||||
"/leaderboard/position": "GET",
|
||||
|
@ -59,6 +60,68 @@ app.add_middleware(
|
|||
)
|
||||
|
||||
|
||||
@app.get("/info")
|
||||
async def get_leadeboard(
|
||||
leaderboard_id: UUID,
|
||||
db_session: Session = Depends(db.yield_db_session),
|
||||
):
|
||||
"""
|
||||
Returns leaderboard info.
|
||||
"""
|
||||
try:
|
||||
leaderboard = actions.get_leaderboard(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,
|
||||
)
|
||||
|
||||
|
||||
@app.get("/leaderboards")
|
||||
async def get_leaderboards(
|
||||
request: Request, db_session: Session = Depends(db.yield_db_session)
|
||||
):
|
||||
"""
|
||||
Returns leaderboard list to which user has access.
|
||||
"""
|
||||
|
||||
token = request.state.token
|
||||
|
||||
try:
|
||||
leaderboards = actions.get_leaderboards(db_session, token)
|
||||
except actions.LeaderboardsResourcesNotFound as e:
|
||||
raise EngineHTTPException(
|
||||
status_code=404,
|
||||
detail="Leaderboards not found.",
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Error while getting leaderboards: {e}")
|
||||
raise EngineHTTPException(status_code=500, detail="Internal server error")
|
||||
|
||||
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,
|
||||
)
|
||||
for leaderboard in leaderboards
|
||||
]
|
||||
|
||||
return results
|
||||
|
||||
|
||||
@app.get("/count/addresses")
|
||||
async def count_addresses(
|
||||
leaderboard_id: UUID,
|
||||
|
@ -155,7 +218,17 @@ async def position(
|
|||
db_session, leaderboard_id, address, window_size, limit, offset
|
||||
)
|
||||
|
||||
return positions
|
||||
results = [
|
||||
data.LeaderboardPosition(
|
||||
address=position.address,
|
||||
score=position.score,
|
||||
rank=position.rank,
|
||||
points_data=position.points_data,
|
||||
)
|
||||
for position in positions
|
||||
]
|
||||
|
||||
return results
|
||||
|
||||
|
||||
@app.get("")
|
||||
|
@ -237,7 +310,7 @@ async def rank(
|
|||
return results
|
||||
|
||||
|
||||
@app.get("/ranks")
|
||||
@app.get("/ranks", response_model=List[data.RanksResponse])
|
||||
async def ranks(
|
||||
leaderboard_id: UUID, db_session: Session = Depends(db.yield_db_session)
|
||||
) -> List[data.RanksResponse]:
|
||||
|
@ -269,15 +342,15 @@ async def ranks(
|
|||
return results
|
||||
|
||||
|
||||
@app.put("/{leaderboard_id}/scores")
|
||||
async def leaderboard(
|
||||
@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,
|
||||
db_session: Session = Depends(db.yield_db_session),
|
||||
):
|
||||
) -> List[data.LeaderboardScore]:
|
||||
"""
|
||||
Put the leaderboard to the database.
|
||||
"""
|
||||
|
@ -328,4 +401,14 @@ async def leaderboard(
|
|||
logger.error(f"Score update failed with error: {e}")
|
||||
raise EngineHTTPException(status_code=500, detail="Score update failed.")
|
||||
|
||||
return leaderboard_points
|
||||
result = [
|
||||
data.LeaderboardScore(
|
||||
leaderboard_id=score["leaderboard_id"],
|
||||
address=score["address"],
|
||||
score=score["score"],
|
||||
points_data=score["points_data"],
|
||||
)
|
||||
for score in leaderboard_points
|
||||
]
|
||||
|
||||
return result
|
||||
|
|
Ładowanie…
Reference in New Issue