kopia lustrzana https://github.com/bugout-dev/moonstream
				
				
				
			Add initial leaderboard manage.
							rodzic
							
								
									f5b85d0b5f
								
							
						
					
					
						commit
						0403b1ba0c
					
				| 
						 | 
				
			
			@ -64,6 +64,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 +220,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 +272,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 +287,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 +312,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 +631,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 +845,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(
 | 
			
		||||
| 
						 | 
				
			
			@ -943,18 +955,79 @@ 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) -> Any:
 | 
			
		||||
    """
 | 
			
		||||
    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)
 | 
			
		||||
        .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
 | 
			
		||||
) -> Any:
 | 
			
		||||
 | 
			
		||||
    """
 | 
			
		||||
    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())
 | 
			
		||||
    )
 | 
			
		||||
 | 
			
		||||
    return leaderboard_scores_changes
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
def get_leaderboard_scores_by_timestamp(
 | 
			
		||||
    db_session: Session,
 | 
			
		||||
    leaderboard_id: uuid.UUID,
 | 
			
		||||
    date: datetime,
 | 
			
		||||
    limit: int,
 | 
			
		||||
    offset: int,
 | 
			
		||||
) -> Any:
 | 
			
		||||
 | 
			
		||||
    """
 | 
			
		||||
    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],
 | 
			
		||||
| 
						 | 
				
			
			@ -1157,30 +1230,105 @@ 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: uuid.UUID,
 | 
			
		||||
):
 | 
			
		||||
    """
 | 
			
		||||
    Create a leaderboard
 | 
			
		||||
    """
 | 
			
		||||
    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
 | 
			
		||||
):
 | 
			
		||||
 | 
			
		||||
    """
 | 
			
		||||
    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}")
 | 
			
		||||
 | 
			
		||||
        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],
 | 
			
		||||
    token: uuid.UUID,
 | 
			
		||||
):
 | 
			
		||||
 | 
			
		||||
    """
 | 
			
		||||
    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
 | 
			
		||||
 | 
			
		||||
    leaderboard = Leaderboard(title=title, description=description)
 | 
			
		||||
    db_session.add(leaderboard)
 | 
			
		||||
    db_session.commit()
 | 
			
		||||
 | 
			
		||||
    return leaderboard.id
 | 
			
		||||
    return leaderboard
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
def get_leaderboard_by_id(db_session: Session, leaderboard_id):
 | 
			
		||||
    """
 | 
			
		||||
    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):
 | 
			
		||||
    """
 | 
			
		||||
    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):
 | 
			
		||||
| 
						 | 
				
			
			@ -1265,8 +1413,10 @@ 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,
 | 
			
		||||
    title: Optional[str] = None,
 | 
			
		||||
    user_id: Optional[uuid.UUID] = None,
 | 
			
		||||
) -> BugoutResource:
 | 
			
		||||
    resource_data: Dict[str, Any] = {
 | 
			
		||||
        "type": LEADERBOARD_RESOURCE_TYPE,
 | 
			
		||||
| 
						 | 
				
			
			@ -1275,19 +1425,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 +1448,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
 | 
			
		||||
| 
						 | 
				
			
			@ -1338,7 +1489,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:
 | 
			
		||||
| 
						 | 
				
			
			@ -1359,7 +1510,7 @@ def check_leaderboard_resource_permissions(
 | 
			
		|||
    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"
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
| 
						 | 
				
			
			@ -355,3 +355,46 @@ class LeaderboardInfoResponse(BaseModel):
 | 
			
		|||
    id: UUID
 | 
			
		||||
    title: str
 | 
			
		||||
    description: Optional[str] = None
 | 
			
		||||
    users_count: int
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
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 LeaderboardUpdatedResponse(BaseModel):
 | 
			
		||||
    id: UUID
 | 
			
		||||
    title: str
 | 
			
		||||
    description: Optional[str] = None
 | 
			
		||||
    resource_id: Optional[UUID] = None
 | 
			
		||||
    created_at: datetime
 | 
			
		||||
    updated_at: datetime
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
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
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
| 
						 | 
				
			
			@ -1,6 +1,7 @@
 | 
			
		|||
"""
 | 
			
		||||
Leaderboard API.
 | 
			
		||||
"""
 | 
			
		||||
from datetime import datetime
 | 
			
		||||
import logging
 | 
			
		||||
from uuid import UUID
 | 
			
		||||
 | 
			
		||||
| 
						 | 
				
			
			@ -37,6 +38,8 @@ leaderboad_whitelist = {
 | 
			
		|||
    "/leaderboard": "GET",
 | 
			
		||||
    "/leaderboard/rank": "GET",
 | 
			
		||||
    "/leaderboard/ranks": "GET",
 | 
			
		||||
    "/leaderboard/docs": "GET",
 | 
			
		||||
    "/leaderboard/openapi.json": "GET",
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
app = FastAPI(
 | 
			
		||||
| 
						 | 
				
			
			@ -69,7 +72,7 @@ async def get_leadeboard(
 | 
			
		|||
    Returns leaderboard info.
 | 
			
		||||
    """
 | 
			
		||||
    try:
 | 
			
		||||
        leaderboard = actions.get_leaderboard(db_session, leaderboard_id)
 | 
			
		||||
        leaderboard = actions.get_leaderboard_info(db_session, leaderboard_id)
 | 
			
		||||
    except NoResultFound as e:
 | 
			
		||||
        raise EngineHTTPException(
 | 
			
		||||
            status_code=404,
 | 
			
		||||
| 
						 | 
				
			
			@ -83,6 +86,8 @@ async def get_leadeboard(
 | 
			
		|||
        id=leaderboard.id,
 | 
			
		||||
        title=leaderboard.title,
 | 
			
		||||
        description=leaderboard.description,
 | 
			
		||||
        users_count=leaderboard.users_count,
 | 
			
		||||
        last_updated=leaderboard.last_update,
 | 
			
		||||
    )
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
| 
						 | 
				
			
			@ -122,6 +127,35 @@ async def get_leaderboards(
 | 
			
		|||
    return results
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
@app.get("/scores/changes")
 | 
			
		||||
async def get_scores_changes(
 | 
			
		||||
    leaderboard_id: UUID,
 | 
			
		||||
    db_session: Session = Depends(db.yield_db_session),
 | 
			
		||||
) -> Any:
 | 
			
		||||
 | 
			
		||||
    """
 | 
			
		||||
    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("/count/addresses", response_model=data.CountAddressesResponse)
 | 
			
		||||
async def count_addresses(
 | 
			
		||||
    leaderboard_id: UUID,
 | 
			
		||||
| 
						 | 
				
			
			@ -271,6 +305,158 @@ async def leaderboard(
 | 
			
		|||
    return result
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
@app.post("", response_model=data.LeaderboardCreatedResponse)
 | 
			
		||||
@app.post("/", response_model=data.LeaderboardCreatedResponse)
 | 
			
		||||
async def create_leaderboard(
 | 
			
		||||
    request: Request,
 | 
			
		||||
    leaderboard: data.LeaderboardCreateRequest,
 | 
			
		||||
    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,
 | 
			
		||||
        title=created_leaderboard.title,
 | 
			
		||||
        description=created_leaderboard.description,
 | 
			
		||||
        resource_id=created_leaderboard.resource_id,
 | 
			
		||||
        created_at=created_leaderboard.created_at,
 | 
			
		||||
        updated_at=created_leaderboard.updated_at,
 | 
			
		||||
    )
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
@app.put("/{leaderboard_id}", response_model=data.LeaderboardUpdatedResponse)
 | 
			
		||||
async def update_leaderboard(
 | 
			
		||||
    request: Request,
 | 
			
		||||
    leaderboard_id: UUID,
 | 
			
		||||
    leaderboard: data.LeaderboardUpdateRequest,
 | 
			
		||||
    db_session: Session = Depends(db.yield_db_session),
 | 
			
		||||
) -> data.LeaderboardUpdatedResponse:
 | 
			
		||||
 | 
			
		||||
    """
 | 
			
		||||
 | 
			
		||||
    Update leaderboard.
 | 
			
		||||
    """
 | 
			
		||||
 | 
			
		||||
    token = request.state.token
 | 
			
		||||
 | 
			
		||||
    access = actions.check_leaderboard_resource_permissions(
 | 
			
		||||
        db_session=db_session,
 | 
			
		||||
        leaderboard_id=leaderboard_id,
 | 
			
		||||
        token=request.state.token,
 | 
			
		||||
    )
 | 
			
		||||
 | 
			
		||||
    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,
 | 
			
		||||
            token=token,
 | 
			
		||||
        )
 | 
			
		||||
    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,
 | 
			
		||||
        title=updated_leaderboard.title,
 | 
			
		||||
        description=updated_leaderboard.description,
 | 
			
		||||
        resource_id=updated_leaderboard.resource_id,
 | 
			
		||||
        created_at=updated_leaderboard.created_at,
 | 
			
		||||
        updated_at=updated_leaderboard.updated_at,
 | 
			
		||||
    )
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
@app.delete("/{leaderboard_id}", response_model=data.LeaderboardDeletedResponse)
 | 
			
		||||
async def delete_leaderboard(
 | 
			
		||||
    request: Request,
 | 
			
		||||
    leaderboard_id: UUID,
 | 
			
		||||
    db_session: Session = Depends(db.yield_db_session),
 | 
			
		||||
) -> data.LeaderboardDeletedResponse:
 | 
			
		||||
 | 
			
		||||
    """
 | 
			
		||||
 | 
			
		||||
    Delete leaderboard.
 | 
			
		||||
    """
 | 
			
		||||
 | 
			
		||||
    token = request.state.token
 | 
			
		||||
 | 
			
		||||
    access = actions.check_leaderboard_resource_permissions(
 | 
			
		||||
        db_session=db_session,
 | 
			
		||||
        leaderboard_id=leaderboard_id,
 | 
			
		||||
        token=request.state.token,
 | 
			
		||||
    )
 | 
			
		||||
 | 
			
		||||
    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,
 | 
			
		||||
        title=deleted_leaderboard.title,
 | 
			
		||||
        description=deleted_leaderboard.description,
 | 
			
		||||
        resource_id=deleted_leaderboard.resource_id,
 | 
			
		||||
        created_at=deleted_leaderboard.created_at,
 | 
			
		||||
        updated_at=deleted_leaderboard.updated_at,
 | 
			
		||||
    )
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
@app.get("/rank", response_model=List[data.LeaderboardPosition])
 | 
			
		||||
async def rank(
 | 
			
		||||
    leaderboard_id: UUID,
 | 
			
		||||
| 
						 | 
				
			
			
 | 
			
		|||
		Ładowanie…
	
		Reference in New Issue