feat(minifigures): add core models for individual minifigure tracking
This commit is contained in:
@@ -0,0 +1,548 @@
|
||||
import logging
|
||||
import traceback
|
||||
from typing import Any, Self, TYPE_CHECKING
|
||||
from uuid import uuid4
|
||||
|
||||
from flask import current_app, url_for
|
||||
|
||||
from .exceptions import NotFoundException, DatabaseException, ErrorException
|
||||
from .parser import parse_minifig
|
||||
from .rebrickable import Rebrickable
|
||||
from .rebrickable_minifigure import RebrickableMinifigure
|
||||
from .set_owner_list import BrickSetOwnerList
|
||||
from .set_purchase_location_list import BrickSetPurchaseLocationList
|
||||
from .set_storage_list import BrickSetStorageList
|
||||
from .set_tag_list import BrickSetTagList
|
||||
from .sql import BrickSQL
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .socket import BrickSocket
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# Individual minifigure (not associated with a set)
|
||||
class IndividualMinifigure(RebrickableMinifigure):
|
||||
# Queries
|
||||
select_query: str = 'individual_minifigure/select/by_id'
|
||||
insert_query: str = 'individual_minifigure/insert'
|
||||
|
||||
# Delete an individual minifigure
|
||||
def delete(self, /) -> None:
|
||||
BrickSQL().executescript(
|
||||
'individual_minifigure/delete',
|
||||
id=self.fields.id
|
||||
)
|
||||
|
||||
# Import an individual minifigure into the database
|
||||
def download(self, socket: 'BrickSocket', data: dict[str, Any], /) -> bool:
|
||||
# Load the minifigure
|
||||
if not self.load(socket, data, from_download=True):
|
||||
return False
|
||||
|
||||
try:
|
||||
# Insert into the database
|
||||
socket.auto_progress(
|
||||
message='Minifigure {figure}: inserting into database'.format(
|
||||
figure=self.fields.figure
|
||||
),
|
||||
increment_total=True,
|
||||
)
|
||||
|
||||
# Generate an UUID for self
|
||||
self.fields.id = str(uuid4())
|
||||
|
||||
# Save the storage
|
||||
storage = BrickSetStorageList.get(
|
||||
data.get('storage', ''),
|
||||
allow_none=True
|
||||
)
|
||||
self.fields.storage = storage.fields.id if storage else None
|
||||
|
||||
# Save the purchase location
|
||||
purchase_location = BrickSetPurchaseLocationList.get(
|
||||
data.get('purchase_location', ''),
|
||||
allow_none=True
|
||||
)
|
||||
self.fields.purchase_location = purchase_location.fields.id if purchase_location else None
|
||||
|
||||
# Save purchase date and price
|
||||
self.fields.purchase_date = data.get('purchase_date', None)
|
||||
self.fields.purchase_price = data.get('purchase_price', None)
|
||||
|
||||
# Save quantity and description
|
||||
self.fields.quantity = int(data.get('quantity', 1))
|
||||
self.fields.description = data.get('description', '')
|
||||
|
||||
# IMPORTANT: Insert rebrickable minifigure FIRST
|
||||
# bricktracker_individual_minifigures has FK to rebrickable_minifigures
|
||||
self.insert_rebrickable_loose()
|
||||
|
||||
# Now insert into bricktracker_individual_minifigures
|
||||
# Use no_defer=True to ensure the insert happens before we insert parts
|
||||
# (parts have a foreign key constraint on this id)
|
||||
self.insert(commit=False, no_defer=True)
|
||||
|
||||
# Save the owners
|
||||
owners: list[str] = list(data.get('owners', []))
|
||||
for id in owners:
|
||||
owner = BrickSetOwnerList.get(id)
|
||||
owner.update_individual_minifigure_state(self, state=True)
|
||||
|
||||
# Save the tags
|
||||
tags: list[str] = list(data.get('tags', []))
|
||||
for id in tags:
|
||||
tag = BrickSetTagList.get(id)
|
||||
tag.update_individual_minifigure_state(self, state=True)
|
||||
|
||||
# Load the parts (elements) for this minifigure
|
||||
if not self.download_parts(socket):
|
||||
return False
|
||||
|
||||
# Commit the transaction to the database
|
||||
socket.auto_progress(
|
||||
message='Minifigure {figure}: writing to the database'.format(
|
||||
figure=self.fields.figure
|
||||
),
|
||||
increment_total=True,
|
||||
)
|
||||
|
||||
BrickSQL().commit()
|
||||
|
||||
# Info
|
||||
logger.info('Minifigure {figure}: imported (id: {id})'.format(
|
||||
figure=self.fields.figure,
|
||||
id=self.fields.id,
|
||||
))
|
||||
|
||||
# Complete
|
||||
socket.complete(
|
||||
message='Minifigure {figure}: imported (<a href="{url}">Go to the minifigure</a>)'.format(
|
||||
figure=self.fields.figure,
|
||||
url=self.url()
|
||||
),
|
||||
download=True
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
socket.fail(
|
||||
message='Error while importing minifigure {figure}: {error}'.format(
|
||||
figure=self.fields.figure,
|
||||
error=e,
|
||||
)
|
||||
)
|
||||
|
||||
logger.debug(traceback.format_exc())
|
||||
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
# Download parts (elements) for this individual minifigure
|
||||
def download_parts(self, socket: 'BrickSocket', /) -> bool:
|
||||
try:
|
||||
# Check if we have cached parts data from load()
|
||||
if hasattr(self, '_cached_parts_response'):
|
||||
response = self._cached_parts_response
|
||||
logger.debug('Using cached parts data from load()')
|
||||
else:
|
||||
# Need to fetch parts data
|
||||
socket.auto_progress(
|
||||
message='Minifigure {figure}: loading parts from Rebrickable'.format(
|
||||
figure=self.fields.figure
|
||||
),
|
||||
increment_total=True,
|
||||
)
|
||||
|
||||
logger.debug('rebrick.lego.get_minifig_elements("{figure}")'.format(
|
||||
figure=self.fields.figure,
|
||||
))
|
||||
|
||||
# Load parts data from Rebrickable API
|
||||
import json
|
||||
from rebrick import lego
|
||||
|
||||
parameters = {
|
||||
'api_key': current_app.config['REBRICKABLE_API_KEY'],
|
||||
'page_size': current_app.config['REBRICKABLE_PAGE_SIZE'],
|
||||
}
|
||||
|
||||
response = json.loads(lego.get_minifig_elements(
|
||||
self.fields.figure,
|
||||
**parameters
|
||||
).read())
|
||||
|
||||
socket.auto_progress(
|
||||
message='Minifigure {figure}: saving parts to database'.format(
|
||||
figure=self.fields.figure
|
||||
),
|
||||
)
|
||||
|
||||
# Insert each part into individual_minifigure_parts table
|
||||
from .rebrickable_part import RebrickablePart
|
||||
|
||||
if 'results' in response:
|
||||
logger.debug('Processing {count} parts for minifigure {figure}'.format(
|
||||
count=len(response["results"]),
|
||||
figure=self.fields.figure
|
||||
))
|
||||
|
||||
for idx, result in enumerate(response['results']):
|
||||
part_num = result['part']['part_num']
|
||||
color_id = result['color']['id']
|
||||
|
||||
logger.debug(
|
||||
'Part {current}/{total}: {part_num} (color: {color_id}, quantity: {quantity})'.format(
|
||||
current=idx+1,
|
||||
total=len(response["results"]),
|
||||
part_num=part_num,
|
||||
color_id=color_id,
|
||||
quantity=result["quantity"]
|
||||
)
|
||||
)
|
||||
|
||||
# Insert rebrickable part data first
|
||||
part_data = RebrickablePart.from_rebrickable(result)
|
||||
logger.debug('Rebrickable part data keys: {keys}'.format(
|
||||
keys=list(part_data.keys())
|
||||
))
|
||||
|
||||
# Insert into rebrickable_parts if not exists
|
||||
BrickSQL().execute(
|
||||
'rebrickable/part/insert',
|
||||
parameters=part_data,
|
||||
commit=False,
|
||||
)
|
||||
|
||||
# Download part image if not using remote images
|
||||
if not current_app.config['USE_REMOTE_IMAGES']:
|
||||
# Create a RebrickablePart instance for image download
|
||||
from .set import BrickSet
|
||||
try:
|
||||
part_instance = RebrickablePart(record=part_data)
|
||||
from .rebrickable_image import RebrickableImage
|
||||
RebrickableImage(
|
||||
BrickSet(), # Dummy set
|
||||
minifigure=self,
|
||||
part=part_instance,
|
||||
).download()
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
'Could not download image for part {part_num}: {error}'.format(
|
||||
part_num=part_num,
|
||||
error=e
|
||||
)
|
||||
)
|
||||
|
||||
# Insert into bricktracker_individual_minifigure_parts
|
||||
individual_part_params = {
|
||||
'id': self.fields.id,
|
||||
'part': part_num,
|
||||
'color': color_id,
|
||||
'spare': result.get('is_spare', False),
|
||||
'quantity': result['quantity'],
|
||||
'element': result.get('element_id'),
|
||||
'rebrickable_inventory': result['id'],
|
||||
}
|
||||
logger.debug('Individual part params: {params}'.format(
|
||||
params=individual_part_params
|
||||
))
|
||||
|
||||
BrickSQL().execute(
|
||||
'individual_minifigure/part/insert',
|
||||
parameters=individual_part_params,
|
||||
commit=False,
|
||||
)
|
||||
|
||||
logger.debug('Successfully inserted all {count} parts'.format(
|
||||
count=len(response["results"])
|
||||
))
|
||||
else:
|
||||
logger.warning('No results in parts response for minifigure {figure}'.format(
|
||||
figure=self.fields.figure
|
||||
))
|
||||
|
||||
# Clean up cached data
|
||||
if hasattr(self, '_cached_parts_response'):
|
||||
delattr(self, '_cached_parts_response')
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
socket.fail(
|
||||
message='Error loading parts for minifigure {figure}: {error}'.format(
|
||||
figure=self.fields.figure,
|
||||
error=e,
|
||||
)
|
||||
)
|
||||
logger.debug(traceback.format_exc())
|
||||
return False
|
||||
|
||||
# Insert the individual minifigure from Rebrickable
|
||||
def insert_rebrickable_loose(self, /) -> None:
|
||||
# Insert the Rebrickable minifigure to the database
|
||||
# Note: We override the parent's insert_rebrickable since we don't have a brickset
|
||||
from .rebrickable_image import RebrickableImage
|
||||
|
||||
# Explicitly build parameters for rebrickable_minifigures insert
|
||||
params = {
|
||||
'figure': self.fields.figure,
|
||||
'number': self.fields.number,
|
||||
'name': self.fields.name,
|
||||
'image': self.fields.image,
|
||||
'number_of_parts': self.fields.number_of_parts,
|
||||
}
|
||||
|
||||
BrickSQL().execute(
|
||||
RebrickableMinifigure.insert_query,
|
||||
parameters=params,
|
||||
commit=False,
|
||||
)
|
||||
|
||||
# Download image locally if not using remote images
|
||||
if not current_app.config['USE_REMOTE_IMAGES']:
|
||||
# Create a dummy BrickSet for RebrickableImage
|
||||
# RebrickableImage checks minifigure first before set, so this works
|
||||
from .set import BrickSet
|
||||
try:
|
||||
RebrickableImage(
|
||||
BrickSet(), # Dummy set - not used since minifigure takes priority
|
||||
minifigure=self,
|
||||
).download()
|
||||
logger.debug('Downloaded image for individual minifigure {figure}'.format(
|
||||
figure=self.fields.figure
|
||||
))
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
'Could not download image for individual minifigure {figure}: {error}'.format(
|
||||
figure=self.fields.figure,
|
||||
error=e
|
||||
)
|
||||
)
|
||||
|
||||
# Load the minifigure from Rebrickable
|
||||
def load(
|
||||
self,
|
||||
socket: 'BrickSocket',
|
||||
data: dict[str, Any],
|
||||
/,
|
||||
*,
|
||||
from_download=False,
|
||||
) -> bool:
|
||||
# Reset the progress
|
||||
socket.progress_count = 0
|
||||
socket.progress_total = 2
|
||||
|
||||
try:
|
||||
# Check if individual minifigures are disabled
|
||||
from flask import current_app
|
||||
if current_app.config.get('DISABLE_INDIVIDUAL_MINIFIGURES', False):
|
||||
raise ErrorException(
|
||||
'Individual minifigures system is disabled. '
|
||||
'Only set-based minifigures can be added.'
|
||||
)
|
||||
|
||||
socket.auto_progress(message='Parsing minifigure number')
|
||||
figure = parse_minifig(str(data['figure']))
|
||||
|
||||
socket.auto_progress(
|
||||
message='Minifigure {figure}: loading from Rebrickable'.format(
|
||||
figure=figure,
|
||||
),
|
||||
)
|
||||
|
||||
logger.debug('rebrick.lego.get_minifig_elements("{figure}")'.format(
|
||||
figure=figure,
|
||||
))
|
||||
|
||||
# Load from Rebrickable using get_minifig_elements
|
||||
# This gives us both minifigure info and parts in one call
|
||||
import json
|
||||
from rebrick import lego
|
||||
|
||||
parameters = {
|
||||
'api_key': current_app.config['REBRICKABLE_API_KEY'],
|
||||
'page_size': current_app.config['REBRICKABLE_PAGE_SIZE'],
|
||||
}
|
||||
|
||||
response = json.loads(lego.get_minifig_elements(
|
||||
figure,
|
||||
**parameters
|
||||
).read())
|
||||
|
||||
# Extract minifigure info from the first part's metadata
|
||||
if 'results' in response and len(response['results']) > 0:
|
||||
first_part = response['results'][0]
|
||||
|
||||
# Build minifigure data from the response
|
||||
self.fields.figure = first_part['set_num']
|
||||
self.fields.number_of_parts = response['count']
|
||||
|
||||
# We need to fetch the proper name and image from get_minifig()
|
||||
# This is a small additional call but gives us the proper minifigure data
|
||||
try:
|
||||
# get_minifig() only needs api_key, not page_size
|
||||
minifig_params = {
|
||||
'api_key': current_app.config['REBRICKABLE_API_KEY']
|
||||
}
|
||||
minifig_response = json.loads(lego.get_minifig(
|
||||
figure,
|
||||
**minifig_params
|
||||
).read())
|
||||
self.fields.name = minifig_response.get('name', "Minifigure {figure}".format(figure=figure))
|
||||
|
||||
# Use the minifig image from get_minifig() - this is the assembled minifig
|
||||
self.fields.image = minifig_response.get('set_img_url')
|
||||
|
||||
# Extract number from figure (e.g., fig-005997 -> 5997)
|
||||
try:
|
||||
self.fields.number = int(figure.split('-')[1])
|
||||
except:
|
||||
self.fields.number = 0
|
||||
|
||||
except Exception as e:
|
||||
logger.warning('Could not fetch minifigure name: {error}'.format(
|
||||
error=e
|
||||
))
|
||||
self.fields.name = "Minifigure {figure}".format(figure=figure)
|
||||
# Try to extract number anyway
|
||||
try:
|
||||
self.fields.number = int(figure.split('-')[1])
|
||||
except:
|
||||
self.fields.number = 0
|
||||
|
||||
# Fallback: try to extract image from first part with element_id
|
||||
self.fields.image = None
|
||||
for result in response['results']:
|
||||
if result.get('element_id') and result['part'].get('part_img_url'):
|
||||
self.fields.image = result['part']['part_img_url']
|
||||
break
|
||||
|
||||
# Store the parts data for later use in download
|
||||
self._cached_parts_response = response
|
||||
else:
|
||||
raise NotFoundException('Minifigure {figure} has no parts in Rebrickable'.format(
|
||||
figure=figure
|
||||
))
|
||||
|
||||
# Download minifigure image during preview if not using remote images
|
||||
if not from_download and not current_app.config['USE_REMOTE_IMAGES'] and self.fields.image:
|
||||
from .rebrickable_image import RebrickableImage
|
||||
from .set import BrickSet
|
||||
try:
|
||||
RebrickableImage(
|
||||
BrickSet(),
|
||||
minifigure=self,
|
||||
).download()
|
||||
logger.debug('Downloaded preview image for minifigure {figure}'.format(
|
||||
figure=self.fields.figure
|
||||
))
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
'Could not download preview image for minifigure {figure}: {error}'.format(
|
||||
figure=self.fields.figure,
|
||||
error=e
|
||||
)
|
||||
)
|
||||
|
||||
socket.emit('MINIFIGURE_LOADED', self.short(
|
||||
from_download=from_download
|
||||
))
|
||||
|
||||
if not from_download:
|
||||
socket.complete(
|
||||
message='Minifigure {figure}: loaded from Rebrickable'.format(
|
||||
figure=self.fields.figure
|
||||
)
|
||||
)
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
# Check if this is the "disabled" error - if so, show cleaner message
|
||||
error_msg = str(e)
|
||||
if 'Individual minifigures system is disabled' in error_msg:
|
||||
socket.fail(message=error_msg)
|
||||
else:
|
||||
socket.fail(
|
||||
message='Could not load the minifigure from Rebrickable: {error}. Data: {data}'.format(
|
||||
error=error_msg,
|
||||
data=data,
|
||||
)
|
||||
)
|
||||
|
||||
if not isinstance(e, (NotFoundException, ErrorException)):
|
||||
logger.debug(traceback.format_exc())
|
||||
|
||||
return False
|
||||
|
||||
# Return a short form of the minifigure
|
||||
def short(self, /, *, from_download: bool = False) -> dict[str, Any]:
|
||||
return {
|
||||
'download': from_download,
|
||||
'image': self.url_for_image(),
|
||||
'name': self.fields.name,
|
||||
'figure': self.fields.figure,
|
||||
}
|
||||
|
||||
# Select an individual minifigure by ID
|
||||
def select_by_id(self, id: str, /) -> Self:
|
||||
# Save the ID parameter
|
||||
self.fields.id = id
|
||||
|
||||
# Import status list here to get metadata columns
|
||||
from .set_status_list import BrickSetStatusList
|
||||
|
||||
# Pass metadata columns to the query (using set tables which now handle all entities)
|
||||
context = {
|
||||
'owners': BrickSetOwnerList.as_columns() if BrickSetOwnerList.list() else '',
|
||||
'statuses': BrickSetStatusList.as_columns(all=True) if BrickSetStatusList.list(all=True) else '',
|
||||
'tags': BrickSetTagList.as_columns() if BrickSetTagList.list() else '',
|
||||
}
|
||||
|
||||
if not self.select(**context):
|
||||
raise NotFoundException(
|
||||
'Individual minifigure with ID {id} was not found in the database'.format(
|
||||
id=id,
|
||||
),
|
||||
)
|
||||
|
||||
return self
|
||||
|
||||
# URL to this individual minifigure instance
|
||||
def url(self, /) -> str:
|
||||
return url_for('individual_minifigure.details', id=self.fields.id)
|
||||
|
||||
# String representation for debugging
|
||||
def __repr__(self, /) -> str:
|
||||
figure = getattr(self.fields, 'figure', 'unknown')
|
||||
name = getattr(self.fields, 'name', 'Unknown')
|
||||
qty = getattr(self.fields, 'quantity', 0)
|
||||
return f'<IndividualMinifigure {figure} "{name}" qty:{qty}>'
|
||||
|
||||
# URL for updating quantity
|
||||
def url_for_quantity(self, /) -> str:
|
||||
return url_for('individual_minifigure.update_quantity', id=self.fields.id)
|
||||
|
||||
# URL for updating description
|
||||
def url_for_description(self, /) -> str:
|
||||
return url_for('individual_minifigure.update_description', id=self.fields.id)
|
||||
|
||||
# Parts
|
||||
def generic_parts(self, /):
|
||||
from .part_list import BrickPartList
|
||||
return BrickPartList().from_individual_minifigure(self)
|
||||
|
||||
# Override from_rebrickable to handle minifigure data
|
||||
@staticmethod
|
||||
def from_rebrickable(data: dict[str, Any], /, **_) -> dict[str, Any]:
|
||||
# Extracting number
|
||||
number = int(str(data['set_num'])[5:])
|
||||
|
||||
return {
|
||||
'figure': str(data['set_num']),
|
||||
'number': int(number),
|
||||
'name': str(data['set_name']),
|
||||
'image': data.get('set_img_url'),
|
||||
'number_of_parts': int(data.get('num_parts', 0)),
|
||||
}
|
||||
@@ -0,0 +1,98 @@
|
||||
import logging
|
||||
from typing import Self
|
||||
|
||||
from .individual_minifigure import IndividualMinifigure
|
||||
from .record_list import BrickRecordList
|
||||
from .set_owner_list import BrickSetOwnerList
|
||||
from .set_status_list import BrickSetStatusList
|
||||
from .set_tag_list import BrickSetTagList
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# Individual minifigures list
|
||||
class IndividualMinifigureList(BrickRecordList[IndividualMinifigure]):
|
||||
# Queries
|
||||
all_query: str = 'individual_minifigure/list/all'
|
||||
instances_by_figure_query: str = 'individual_minifigure/select/instances_by_figure'
|
||||
using_storage_query: str = 'individual_minifigure/list/using_storage'
|
||||
using_purchase_location_query: str = 'individual_minifigure/list/using_purchase_location'
|
||||
without_storage_query: str = 'individual_minifigure/list/without_storage'
|
||||
|
||||
def __init__(self, /):
|
||||
super().__init__()
|
||||
|
||||
# Load all individual minifigures
|
||||
def all(self, /) -> Self:
|
||||
# Prepare context with metadata columns
|
||||
context = {
|
||||
'owners': BrickSetOwnerList.as_columns() if BrickSetOwnerList.list() else 'NULL AS "no_owners"',
|
||||
'statuses': BrickSetStatusList.as_columns(all=True) if BrickSetStatusList.list(all=True) else 'NULL AS "no_statuses"',
|
||||
'tags': BrickSetTagList.as_columns() if BrickSetTagList.list() else 'NULL AS "no_tags"',
|
||||
}
|
||||
|
||||
self.list(override_query=self.all_query, **context)
|
||||
return self
|
||||
|
||||
# Load all individual instances of a specific minifigure figure
|
||||
def instances_by_figure(self, figure: str, /) -> Self:
|
||||
self.fields.figure = figure
|
||||
|
||||
# Prepare context with metadata columns (using consolidated metadata tables)
|
||||
context = {
|
||||
'owners': BrickSetOwnerList.as_columns() if BrickSetOwnerList.list() else 'NULL AS "no_owners"',
|
||||
'statuses': BrickSetStatusList.as_columns(all=True) if BrickSetStatusList.list(all=True) else 'NULL AS "no_statuses"',
|
||||
'tags': BrickSetTagList.as_columns() if BrickSetTagList.list() else 'NULL AS "no_tags"',
|
||||
}
|
||||
|
||||
# Load the instances from the database
|
||||
self.list(override_query=self.instances_by_figure_query, **context)
|
||||
|
||||
return self
|
||||
|
||||
# Load all individual minifigures using a specific storage
|
||||
def using_storage(self, storage: 'BrickSetStorage', /) -> Self:
|
||||
# Save the storage parameter
|
||||
self.fields.storage = storage.fields.id
|
||||
|
||||
# Load the minifigures from the database
|
||||
self.list(override_query=self.using_storage_query)
|
||||
|
||||
return self
|
||||
|
||||
# Load all individual minifigures using a specific purchase location
|
||||
def using_purchase_location(self, purchase_location: 'BrickSetPurchaseLocation', /) -> Self:
|
||||
# Save the purchase location parameter
|
||||
self.fields.purchase_location = purchase_location.fields.id
|
||||
|
||||
# Load the minifigures from the database
|
||||
self.list(override_query=self.using_purchase_location_query)
|
||||
|
||||
return self
|
||||
|
||||
# Load all individual minifigures without storage
|
||||
def without_storage(self, /) -> Self:
|
||||
# Load minifigures with no storage
|
||||
self.list(override_query=self.without_storage_query)
|
||||
|
||||
return self
|
||||
|
||||
# Base individual minifigure list
|
||||
def list(
|
||||
self,
|
||||
/,
|
||||
*,
|
||||
override_query: str | None = None,
|
||||
order: str | None = None,
|
||||
limit: int | None = None,
|
||||
**context,
|
||||
) -> None:
|
||||
# Load the individual minifigures from the database
|
||||
for record in super().select(
|
||||
override_query=override_query,
|
||||
order=order,
|
||||
limit=limit,
|
||||
**context
|
||||
):
|
||||
individual_minifigure = IndividualMinifigure(record=record)
|
||||
self.records.append(individual_minifigure)
|
||||
Reference in New Issue
Block a user