14020438663

Committed 23 Mar 2025 03:31PM UTC coverage: 72.696% (+1.1%) from 71.579%

Build # 14020438663

Build Type

push

github

Committed by

web-flow

Commit Message

Add Loki, fix some bugs (#97)

Run Details

10 of 11 new or added lines in 3 files covered. (90.91%)

1 existing line in 1 file now uncovered.

213 of 293 relevant lines covered (72.7%)

0.73 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

73.39

/src/sabogaapi/api_v1/models.py

"""Beanie database models."""

import datetime
from typing import Annotated, List

from beanie import Document, Indexed
from pydantic import BaseModel


class BoardgameComparison(BaseModel):
    bgg_id: int
    name: str = ""
    description: str | None = None
    image_url: str | None = None
    thumbnail_url: str | None = None
    year_published: int | None = None
    minplayers: int | None = None
    maxplayers: int | None = None
    playingtime: int | None = None
    minplaytime: int | None = None
    maxplaytime: int | None = None
    categories: List["Category"] = []
    designers: List["Designer"] = []
    mechanics: List["Mechanic"] = []
    bgg_rank: int
    bgg_rank_change: int
    bgg_geek_rating: float
    bgg_geek_rating_change: float
    bgg_average_rating: float
    bgg_average_rating_change: float


class BoardgameWithHistoricalData(BaseModel):
    bgg_id: int
    name: str = ""
    description: str | None = None
    image_url: str | None = None
    thumbnail_url: str | None = None
    year_published: int | None = None
    minplayers: int | None = None
    maxplayers: int | None = None
    playingtime: int | None = None
    minplaytime: int | None = None
    maxplaytime: int | None = None
    categories: List["Category"] = []
    designers: List["Designer"] = []
    mechanics: List["Mechanic"] = []
    bgg_rank: int
    bgg_geek_rating: float
    bgg_average_rating: float
    bgg_rank_history: List["RankHistory"]


class RankHistory(BaseModel):
    date: datetime.datetime
    bgg_rank: int | None
    bgg_geek_rating: float | None
    bgg_average_rating: float | None


class Category(BaseModel):
    name: str
    bgg_id: int


class Mechanic(BaseModel):
    name: str
    bgg_id: int


class Designer(BaseModel):
    name: str
    bgg_id: int


class Boardgame(Document):
    bgg_id: Annotated[int, Indexed(unique=True)]
    name: str = ""
    description: str | None = None
    image_url: str | None = None
    thumbnail_url: str | None = None
    year_published: int | None = None
    minplayers: int | None = None
    maxplayers: int | None = None
    playingtime: int | None = None
    minplaytime: int | None = None
    maxplaytime: int | None = None
    categories: List[Category] = []
    mechanics: List[Mechanic] = []
    designers: List[Designer] = []
    bgg_rank_history: List["RankHistory"] = []

    @staticmethod
    async def get_top_ranked_boardgames(
        date: datetime.datetime,
        compare_to: datetime.datetime,
        page: int = 1,
        page_size: int = 10,
    ) -> List[BoardgameComparison]:
        pipeline = [
            {
                "$lookup": {
                    "from": "categories",  # The collection name of Category
                    "localField": "category_ids",  # The field in Boardgame
                    "foreignField": "_id",  # The matching field in Category
                    "as": "categories",  # Output field
                }
            },
            {
                "$addFields": {
                    "current_rank_data": {
                        "$arrayElemAt": [
                            {
                                "$filter": {
                                    "input": "$bgg_rank_history",
                                    "as": "history",
                                    "cond": {"$lte": ["$$history.date", date]},
                                }
                            },
                            -1,
                        ]
                    }
                }
            },
            {"$match": {"current_rank_data.bgg_rank": {"$ne": None}}},
            {"$sort": {"current_rank_data.bgg_rank": 1}},
            {"$skip": (page - 1) * page_size},
            {"$limit": page_size},
            {
                "$addFields": {
                    "previous_rank_data": {
                        "$arrayElemAt": [
                            {
                                "$filter": {
                                    "input": "$bgg_rank_history",
                                    "as": "history",
                                    "cond": {"$lte": ["$$history.date", compare_to]},
                                }
                            },
                            -1,
                        ]
                    },
                }
            },
            {
                "$addFields": {
                    "bgg_rank": "$current_rank_data.bgg_rank",
                    "bgg_rank_change": {
                        "$ifNull": [
                            {
                                "$subtract": [
                                    "$previous_rank_data.bgg_rank",
                                    "$current_rank_data.bgg_rank",
                                ]
                            },
                            0,
                        ]
                    },
                    "bgg_geek_rating": "$current_rank_data.bgg_geek_rating",
                    "bgg_geek_rating_change": {
                        "$ifNull": [
                            {
                                "$subtract": [
                                    "$current_rank_data.bgg_geek_rating",
                                    "$previous_rank_data.bgg_geek_rating",
                                ]
                            },
                            0,
                        ]
                    },
                    "bgg_average_rating": "$current_rank_data.bgg_average_rating",
                    "bgg_average_rating_change": {
                        "$ifNull": [
                            {
                                "$subtract": [
                                    "$current_rank_data.bgg_average_rating",
                                    "$previous_rank_data.bgg_average_rating",
                                ]
                            },
                            0,
                        ]
                    },
                    "categories": "$categories",
                }
            },
        ]

        rank_data = await Boardgame.aggregate(
            aggregation_pipeline=pipeline, projection_model=BoardgameComparison
        ).to_list()

        return rank_data

    @staticmethod
    async def get_boardgame_with_historical_data(
        bgg_id: int,
        start_date: datetime.datetime,
        end_date: datetime.datetime,
        mode: str,
    ) -> BoardgameWithHistoricalData | None:
        date_diff = (end_date - start_date).days
        if mode == "auto":
            if date_diff <= 30:
                mode = "daily"
            elif date_diff <= 180:
                mode = "weekly"
            else:
                mode = "yearly"

        pipeline = [
            {"$match": {"bgg_id": bgg_id}},  # Match board game by ID
            {"$unwind": "$bgg_rank_history"},  # Unwind the bgg_rank_history array
            {
                "$match": {  # Filter rank history by date range
                    "bgg_rank_history.date": {"$gte": start_date, "$lte": end_date}
                }
            },
            {"$sort": {"bgg_rank_history.date": 1}},  # Sort by date (ascending)
            {
                "$group": {
                    "_id": "$bgg_id",
                    "latest_rank": {
                        "$last": "$bgg_rank_history"
                    },  # Get latest rank entry
                    "bgg_rank_history": {
                        "$push": "$bgg_rank_history"
                    },  # Keep all history
                    "doc": {"$first": "$$ROOT"},  # Store the full document
                }
            },
            {
                "$project": {
                    "_id": 0,
                    "bgg_id": "$_id",
                    "bgg_rank": "$latest_rank.bgg_rank",
                    "bgg_geek_rating": "$latest_rank.bgg_geek_rating",
                    "bgg_average_rating": "$latest_rank.bgg_average_rating",
                    "doc": 1,  # Keep full document data
                    "bgg_rank_history": 1,
                }
            },
            {"$replaceRoot": {"newRoot": {"$mergeObjects": ["$doc", "$$ROOT"]}}},
        ]

        result = await Boardgame.aggregate(
            aggregation_pipeline=pipeline, projection_model=BoardgameWithHistoricalData
        ).to_list()

        if not result:
            return None

        boardgame_data = result[0]

        # Apply mode filtering to bgg_rank_history
        history = boardgame_data.bgg_rank_history
        if mode == "weekly":
            history = history[::7]  # Every 7th entry
        elif mode == "yearly":
            seen_years = set()
            yearly_history = []
            for entry in history:
                year = entry.date.year
                if year not in seen_years:
                    yearly_history.append(entry)
                    seen_years.add(year)
            history = yearly_history

        boardgame_data.bgg_rank_history = history
        return boardgame_data

    class Settings:
        name = "boardgames"

1	"""Beanie database models."""
2
3	import datetime	1✔
4	from typing import Annotated, List	1✔
5
6	from beanie import Document, Indexed	1✔
7	from pydantic import BaseModel	1✔
8
9
10	class BoardgameComparison(BaseModel):	1✔
11	bgg_id: int	1✔
12	name: str = ""	1✔
13	description: str \| None = None	1✔
14	image_url: str \| None = None	1✔
15	thumbnail_url: str \| None = None	1✔
16	year_published: int \| None = None	1✔
17	minplayers: int \| None = None	1✔
18	maxplayers: int \| None = None	1✔
19	playingtime: int \| None = None	1✔
20	minplaytime: int \| None = None	1✔
21	maxplaytime: int \| None = None	1✔
22	categories: List["Category"] = []	1✔
23	designers: List["Designer"] = []	1✔
24	mechanics: List["Mechanic"] = []	1✔
25	bgg_rank: int	1✔
26	bgg_rank_change: int	1✔
27	bgg_geek_rating: float	1✔
28	bgg_geek_rating_change: float	1✔
29	bgg_average_rating: float	1✔
30	bgg_average_rating_change: float	1✔
31
32
33	class BoardgameWithHistoricalData(BaseModel):	1✔
34	bgg_id: int	1✔
35	name: str = ""	1✔
36	description: str \| None = None	1✔
37	image_url: str \| None = None	1✔
38	thumbnail_url: str \| None = None	1✔
39	year_published: int \| None = None	1✔
40	minplayers: int \| None = None	1✔
41	maxplayers: int \| None = None	1✔
42	playingtime: int \| None = None	1✔
43	minplaytime: int \| None = None	1✔
44	maxplaytime: int \| None = None	1✔
45	categories: List["Category"] = []	1✔
46	designers: List["Designer"] = []	1✔
47	mechanics: List["Mechanic"] = []	1✔
48	bgg_rank: int	1✔
49	bgg_geek_rating: float	1✔
50	bgg_average_rating: float	1✔
51	bgg_rank_history: List["RankHistory"]	1✔
52
53
54	class RankHistory(BaseModel):	1✔
55	date: datetime.datetime	1✔
56	bgg_rank: int \| None	1✔
57	bgg_geek_rating: float \| None	1✔
58	bgg_average_rating: float \| None	1✔
59
60
61	class Category(BaseModel):	1✔
62	name: str	1✔
63	bgg_id: int	1✔
64
65
66	class Mechanic(BaseModel):	1✔
67	name: str	1✔
68	bgg_id: int	1✔
69
70
71	class Designer(BaseModel):	1✔
72	name: str	1✔
73	bgg_id: int	1✔
74
75
76	class Boardgame(Document):	1✔
77	bgg_id: Annotated[int, Indexed(unique=True)]	1✔
78	name: str = ""	1✔
79	description: str \| None = None	1✔
80	image_url: str \| None = None	1✔
81	thumbnail_url: str \| None = None	1✔
82	year_published: int \| None = None	1✔
83	minplayers: int \| None = None	1✔
84	maxplayers: int \| None = None	1✔
85	playingtime: int \| None = None	1✔
86	minplaytime: int \| None = None	1✔
87	maxplaytime: int \| None = None	1✔
88	categories: List[Category] = []	1✔
89	mechanics: List[Mechanic] = []	1✔
90	designers: List[Designer] = []	1✔
91	bgg_rank_history: List["RankHistory"] = []	1✔
92
93	@staticmethod	1✔
94	async def get_top_ranked_boardgames(	1✔
95	date: datetime.datetime,
96	compare_to: datetime.datetime,
97	page: int = 1,
98	page_size: int = 10,
99	) -> List[BoardgameComparison]:
100	pipeline = [	×
101	{
102	"$lookup": {
103	"from": "categories", # The collection name of Category
104	"localField": "category_ids", # The field in Boardgame
105	"foreignField": "_id", # The matching field in Category
106	"as": "categories", # Output field
107	}
108	},
109	{
110	"$addFields": {
111	"current_rank_data": {
112	"$arrayElemAt": [
113	{
114	"$filter": {
115	"input": "$bgg_rank_history",
116	"as": "history",
117	"cond": {"$lte": ["$$history.date", date]},
118	}
119	},
120	-1,
121	]
122	}
123	}
124	},
125	{"$match": {"current_rank_data.bgg_rank": {"$ne": None}}},
126	{"$sort": {"current_rank_data.bgg_rank": 1}},
127	{"$skip": (page - 1) * page_size},
128	{"$limit": page_size},
129	{
130	"$addFields": {
131	"previous_rank_data": {
132	"$arrayElemAt": [
133	{
134	"$filter": {
135	"input": "$bgg_rank_history",
136	"as": "history",
137	"cond": {"$lte": ["$$history.date", compare_to]},
138	}
139	},
140	-1,
141	]
142	},
143	}
144	},
145	{
146	"$addFields": {
147	"bgg_rank": "$current_rank_data.bgg_rank",
148	"bgg_rank_change": {
149	"$ifNull": [
150	{
151	"$subtract": [
152	"$previous_rank_data.bgg_rank",
153	"$current_rank_data.bgg_rank",
154	]
155	},
156	0,
157	]
158	},
159	"bgg_geek_rating": "$current_rank_data.bgg_geek_rating",
160	"bgg_geek_rating_change": {
161	"$ifNull": [
162	{
163	"$subtract": [
164	"$current_rank_data.bgg_geek_rating",
165	"$previous_rank_data.bgg_geek_rating",
166	]
167	},
168	0,
169	]
170	},
171	"bgg_average_rating": "$current_rank_data.bgg_average_rating",
172	"bgg_average_rating_change": {
173	"$ifNull": [
174	{
175	"$subtract": [
176	"$current_rank_data.bgg_average_rating",
177	"$previous_rank_data.bgg_average_rating",
178	]
179	},
180	0,
181	]
182	},
183	"categories": "$categories",
184	}
185	},
186	]
187
188	rank_data = await Boardgame.aggregate(	×
189	aggregation_pipeline=pipeline, projection_model=BoardgameComparison
190	).to_list()
191
192	return rank_data	×
193
194	@staticmethod	1✔
195	async def get_boardgame_with_historical_data(	1✔
196	bgg_id: int,
197	start_date: datetime.datetime,
198	end_date: datetime.datetime,
199	mode: str,
200	) -> BoardgameWithHistoricalData \| None:
201	date_diff = (end_date - start_date).days	×
202	if mode == "auto":	×
203	if date_diff <= 30:	×
204	mode = "daily"	×
205	elif date_diff <= 180:	×
206	mode = "weekly"	×
207	else:
208	mode = "yearly"	×
209
210	pipeline = [	×
211	{"$match": {"bgg_id": bgg_id}}, # Match board game by ID
212	{"$unwind": "$bgg_rank_history"}, # Unwind the bgg_rank_history array
213	{
214	"$match": { # Filter rank history by date range
215	"bgg_rank_history.date": {"$gte": start_date, "$lte": end_date}
216	}
217	},
218	{"$sort": {"bgg_rank_history.date": 1}}, # Sort by date (ascending)
219	{
220	"$group": {
221	"_id": "$bgg_id",
222	"latest_rank": {
223	"$last": "$bgg_rank_history"
224	}, # Get latest rank entry
225	"bgg_rank_history": {
226	"$push": "$bgg_rank_history"
227	}, # Keep all history
228	"doc": {"$first": "$$ROOT"}, # Store the full document
229	}
230	},
231	{
232	"$project": {
233	"_id": 0,
234	"bgg_id": "$_id",
235	"bgg_rank": "$latest_rank.bgg_rank",
236	"bgg_geek_rating": "$latest_rank.bgg_geek_rating",
237	"bgg_average_rating": "$latest_rank.bgg_average_rating",
238	"doc": 1, # Keep full document data
239	"bgg_rank_history": 1,
240	}
241	},
242	{"$replaceRoot": {"newRoot": {"$mergeObjects": ["$doc", "$$ROOT"]}}},
243	]
244
245	result = await Boardgame.aggregate(	×
246	aggregation_pipeline=pipeline, projection_model=BoardgameWithHistoricalData
247	).to_list()
248
UNCOV 249	if not result:	×
250	return None	×
251
252	boardgame_data = result[0]	×
253
254	# Apply mode filtering to bgg_rank_history
255	history = boardgame_data.bgg_rank_history	×
256	if mode == "weekly":	×
257	history = history[::7] # Every 7th entry	×
258	elif mode == "yearly":	×
259	seen_years = set()	×
260	yearly_history = []	×
261	for entry in history:	×
262	year = entry.date.year	×
263	if year not in seen_years:	×
264	yearly_history.append(entry)	×
265	seen_years.add(year)	×
266	history = yearly_history	×
267
268	boardgame_data.bgg_rank_history = history	×
269	return boardgame_data	×
270
271	class Settings:	1✔
272	name = "boardgames"	1✔

Ruberhauptmann / saboga-api / 14020438663

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous