Create a Metadata object as a base for checkboxes
This commit is contained in:
parent
a3cb409749
commit
4e88b33c1f
197
bricktracker/metadata.py
Normal file
197
bricktracker/metadata.py
Normal file
@ -0,0 +1,197 @@
|
||||
import logging
|
||||
from sqlite3 import Row
|
||||
from typing import Any, Self, TYPE_CHECKING
|
||||
from uuid import uuid4
|
||||
|
||||
from flask import url_for
|
||||
|
||||
from .exceptions import DatabaseException, ErrorException, NotFoundException
|
||||
from .record import BrickRecord
|
||||
from .sql import BrickSQL
|
||||
if TYPE_CHECKING:
|
||||
from .set import BrickSet
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# Lego set metadata (customizable list of entries that can be checked)
|
||||
class BrickMetadata(BrickRecord):
|
||||
kind: str
|
||||
prefix: str
|
||||
|
||||
# Set state endpoint
|
||||
set_state_endpoint: str
|
||||
|
||||
# Queries
|
||||
delete_query: str
|
||||
insert_query: str
|
||||
select_query: str
|
||||
update_field_query: str
|
||||
update_set_state_query: str
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
/,
|
||||
*,
|
||||
record: Row | dict[str, Any] | None = None,
|
||||
):
|
||||
super().__init__()
|
||||
|
||||
# Ingest the record if it has one
|
||||
if record is not None:
|
||||
self.ingest(record)
|
||||
|
||||
# SQL column name
|
||||
def as_column(self, /) -> str:
|
||||
return '{prefix}_{id}'.format(id=self.fields.id, prefix=self.prefix)
|
||||
|
||||
# HTML dataset name
|
||||
def as_dataset(self, /) -> str:
|
||||
return '{id}'.format(
|
||||
id=self.as_column().replace('_', '-')
|
||||
)
|
||||
|
||||
# Delete from database
|
||||
def delete(self, /) -> None:
|
||||
BrickSQL().executescript(
|
||||
self.delete_query,
|
||||
id=self.fields.id,
|
||||
)
|
||||
|
||||
# Insert into database
|
||||
def insert(self, /, **context) -> None:
|
||||
self.safe()
|
||||
|
||||
# Generate an ID for the metadata (with underscores to make it
|
||||
# column name friendly)
|
||||
self.fields.id = str(uuid4()).replace('-', '_')
|
||||
|
||||
BrickSQL().executescript(
|
||||
self.insert_query,
|
||||
id=self.fields.id,
|
||||
name=self.fields.safe_name,
|
||||
**context
|
||||
)
|
||||
|
||||
# Rename the entry
|
||||
def rename(self, /) -> None:
|
||||
self.safe()
|
||||
|
||||
self.update_field('name', value=self.fields.name)
|
||||
|
||||
# Make the name "safe"
|
||||
# Security: eh.
|
||||
def safe(self, /) -> None:
|
||||
# Prevent self-ownage with accidental quote escape
|
||||
self.fields.safe_name = self.fields.name.replace("'", "''")
|
||||
|
||||
# URL to change the selected state of this metadata item for a set
|
||||
def url_for_set_state(self, id: str, /) -> str:
|
||||
return url_for(
|
||||
self.set_state_endpoint,
|
||||
id=id,
|
||||
metadata_id=self.fields.id
|
||||
)
|
||||
|
||||
# Select a specific checkbox (with an id)
|
||||
def select_specific(self, id: str, /) -> Self:
|
||||
# Save the parameters to the fields
|
||||
self.fields.id = id
|
||||
|
||||
# Load from database
|
||||
if not self.select():
|
||||
raise NotFoundException(
|
||||
'{kind} with ID {id} was not found in the database'.format(
|
||||
kind=self.kind.capitalize(),
|
||||
id=self.fields.id,
|
||||
),
|
||||
)
|
||||
|
||||
return self
|
||||
|
||||
# Update a field
|
||||
def update_field(
|
||||
self,
|
||||
field: str,
|
||||
/,
|
||||
*,
|
||||
json: Any | None = None,
|
||||
value: Any | None = None
|
||||
) -> Any:
|
||||
if value is None:
|
||||
value = json.get('value', None) # type: ignore
|
||||
|
||||
if value is None:
|
||||
raise ErrorException('"{field}" of a {kind} cannot be set to an empty value'.format( # noqa: E501
|
||||
field=field,
|
||||
kind=self.kind
|
||||
))
|
||||
|
||||
if field == 'id' or not hasattr(self.fields, field):
|
||||
raise NotFoundException('"{field}" is not a field of a {kind}'.format( # noqa: E501
|
||||
kind=self.kind,
|
||||
field=field
|
||||
))
|
||||
|
||||
parameters = self.sql_parameters()
|
||||
parameters['value'] = value
|
||||
|
||||
# Update the status
|
||||
rows, _ = BrickSQL().execute_and_commit(
|
||||
self.update_field_query,
|
||||
parameters=parameters,
|
||||
field=field,
|
||||
)
|
||||
|
||||
if rows != 1:
|
||||
raise DatabaseException('Could not update the field "{field}" for {kind} {name} ({id})'.format( # noqa: E501
|
||||
field=field,
|
||||
kind=self.kind,
|
||||
name=self.fields.name,
|
||||
id=self.fields.id,
|
||||
))
|
||||
|
||||
# Info
|
||||
logger.info('{kind} "{name}" ({id}): field "{field}" changed to "{value}"'.format( # noqa: E501
|
||||
kind=self.kind.capitalize(),
|
||||
name=self.fields.name,
|
||||
id=self.fields.id,
|
||||
field=field,
|
||||
value=value,
|
||||
))
|
||||
|
||||
return value
|
||||
|
||||
# Update the selected state of this metadata item for a set
|
||||
def update_set_state(self, brickset: 'BrickSet', json: Any | None) -> Any:
|
||||
state: bool = json.get('value', False) # type: ignore
|
||||
|
||||
parameters = self.sql_parameters()
|
||||
parameters['set_id'] = brickset.fields.id
|
||||
parameters['state'] = state
|
||||
|
||||
# Update the status
|
||||
rows, _ = BrickSQL().execute_and_commit(
|
||||
self.update_set_state_query,
|
||||
parameters=parameters,
|
||||
name=self.as_column(),
|
||||
)
|
||||
|
||||
if rows != 1:
|
||||
raise DatabaseException('Could not update the {kind} "{name}" state for set {set} ({id})'.format( # noqa: E501
|
||||
kind=self.kind,
|
||||
name=self.fields.name,
|
||||
set=brickset.fields.set,
|
||||
id=brickset.fields.id,
|
||||
))
|
||||
|
||||
# Info
|
||||
logger.info('{kind} "{name}" state change to "{state}" for set {set} ({id})'.format( # noqa: E501
|
||||
kind=self.kind,
|
||||
name=self.fields.name,
|
||||
state=state,
|
||||
set=brickset.fields.set,
|
||||
id=brickset.fields.id,
|
||||
))
|
||||
|
||||
return state
|
@ -5,11 +5,10 @@ from uuid import uuid4
|
||||
|
||||
from flask import current_app, url_for
|
||||
|
||||
from .exceptions import DatabaseException, NotFoundException
|
||||
from .exceptions import NotFoundException
|
||||
from .minifigure_list import BrickMinifigureList
|
||||
from .part_list import BrickPartList
|
||||
from .rebrickable_set import RebrickableSet
|
||||
from .set_checkbox import BrickSetCheckbox
|
||||
from .set_checkbox_list import BrickSetCheckboxList
|
||||
from .sql import BrickSQL
|
||||
if TYPE_CHECKING:
|
||||
@ -172,30 +171,6 @@ class BrickSet(RebrickableSet):
|
||||
|
||||
return self
|
||||
|
||||
# Update a status
|
||||
def update_status(
|
||||
self,
|
||||
checkbox: BrickSetCheckbox,
|
||||
status: bool,
|
||||
/
|
||||
) -> None:
|
||||
parameters = self.sql_parameters()
|
||||
parameters['status'] = status
|
||||
|
||||
# Update the status
|
||||
rows, _ = BrickSQL().execute_and_commit(
|
||||
'set/update/status',
|
||||
parameters=parameters,
|
||||
name=checkbox.as_column(),
|
||||
)
|
||||
|
||||
if rows != 1:
|
||||
raise DatabaseException('Could not update the status "{status}" for set {set} ({id})'.format( # noqa: E501
|
||||
status=checkbox.fields.name,
|
||||
set=self.fields.set,
|
||||
id=self.fields.id,
|
||||
))
|
||||
|
||||
# Self url
|
||||
def url(self, /) -> str:
|
||||
return url_for('set.details', id=self.fields.id)
|
||||
|
@ -1,139 +1,39 @@
|
||||
from sqlite3 import Row
|
||||
from typing import Any, Self
|
||||
from uuid import uuid4
|
||||
from typing import Self
|
||||
|
||||
from flask import url_for
|
||||
|
||||
from .exceptions import DatabaseException, ErrorException, NotFoundException
|
||||
from .record import BrickRecord
|
||||
from .sql import BrickSQL
|
||||
from .exceptions import ErrorException
|
||||
from .metadata import BrickMetadata
|
||||
|
||||
|
||||
# Lego set checkbox
|
||||
class BrickSetCheckbox(BrickRecord):
|
||||
class BrickSetCheckbox(BrickMetadata):
|
||||
kind: str = 'checkbox'
|
||||
prefix: str = 'status'
|
||||
|
||||
# Set state endpoint
|
||||
set_state_endpoint: str = 'set.update_status'
|
||||
|
||||
# Queries
|
||||
delete_query: str = 'checkbox/delete'
|
||||
insert_query: str = 'checkbox/insert'
|
||||
select_query: str = 'checkbox/select'
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
/,
|
||||
*,
|
||||
record: Row | dict[str, Any] | None = None,
|
||||
):
|
||||
super().__init__()
|
||||
|
||||
# Ingest the record if it has one
|
||||
if record is not None:
|
||||
self.ingest(record)
|
||||
|
||||
# SQL column name
|
||||
def as_column(self) -> str:
|
||||
return 'status_{id}'.format(id=self.fields.id)
|
||||
|
||||
# HTML dataset name
|
||||
def as_dataset(self) -> str:
|
||||
return '{id}'.format(
|
||||
id=self.as_column().replace('_', '-')
|
||||
)
|
||||
|
||||
# Delete from database
|
||||
def delete(self) -> None:
|
||||
BrickSQL().executescript(
|
||||
'checkbox/delete',
|
||||
id=self.fields.id,
|
||||
)
|
||||
update_field_query: str = 'checkbox/update/field'
|
||||
update_set_state_query: str = 'set/update/status'
|
||||
|
||||
# Grab data from a form
|
||||
def from_form(self, form: dict[str, str]) -> Self:
|
||||
def from_form(self, form: dict[str, str], /) -> Self:
|
||||
name = form.get('name', None)
|
||||
grid = form.get('grid', None)
|
||||
|
||||
if name is None or name == '':
|
||||
raise ErrorException('Checkbox name cannot be empty')
|
||||
|
||||
# Security: eh.
|
||||
# Prevent self-ownage with accidental quote escape
|
||||
self.fields.name = name
|
||||
self.fields.safe_name = self.fields.name.replace("'", "''")
|
||||
self.fields.displayed_on_grid = grid == 'on'
|
||||
|
||||
return self
|
||||
|
||||
# Insert into database
|
||||
def insert(self, **_) -> None:
|
||||
# Generate an ID for the checkbox (with underscores to make it
|
||||
# column name friendly)
|
||||
self.fields.id = str(uuid4()).replace('-', '_')
|
||||
|
||||
BrickSQL().executescript(
|
||||
'checkbox/add',
|
||||
id=self.fields.id,
|
||||
name=self.fields.safe_name,
|
||||
def insert(self, /, **_) -> None:
|
||||
super().insert(
|
||||
displayed_on_grid=self.fields.displayed_on_grid
|
||||
)
|
||||
|
||||
# Rename the checkbox
|
||||
def rename(self, /) -> None:
|
||||
# Update the name
|
||||
rows, _ = BrickSQL().execute_and_commit(
|
||||
'checkbox/update/name',
|
||||
parameters=self.sql_parameters(),
|
||||
)
|
||||
|
||||
if rows != 1:
|
||||
raise DatabaseException('Could not update the name for checkbox {name} ({id})'.format( # noqa: E501
|
||||
name=self.fields.name,
|
||||
id=self.fields.id,
|
||||
))
|
||||
|
||||
# URL to change the status
|
||||
def status_url(self, id: str) -> str:
|
||||
return url_for(
|
||||
'set.update_status',
|
||||
id=id,
|
||||
checkbox_id=self.fields.id
|
||||
)
|
||||
|
||||
# Select a specific checkbox (with an id)
|
||||
def select_specific(self, id: str, /) -> Self:
|
||||
# Save the parameters to the fields
|
||||
self.fields.id = id
|
||||
|
||||
# Load from database
|
||||
if not self.select():
|
||||
raise NotFoundException(
|
||||
'Checkbox with ID {id} was not found in the database'.format(
|
||||
id=self.fields.id,
|
||||
),
|
||||
)
|
||||
|
||||
return self
|
||||
|
||||
# Update a status
|
||||
def update_status(
|
||||
self,
|
||||
name: str,
|
||||
status: bool,
|
||||
/
|
||||
) -> None:
|
||||
if not hasattr(self.fields, name) or name in ['id', 'name']:
|
||||
raise NotFoundException('{name} is not a field of a checkbox'.format( # noqa: E501
|
||||
name=name
|
||||
))
|
||||
|
||||
parameters = self.sql_parameters()
|
||||
parameters['status'] = status
|
||||
|
||||
# Update the status
|
||||
rows, _ = BrickSQL().execute_and_commit(
|
||||
'checkbox/update/status',
|
||||
parameters=parameters,
|
||||
name=name,
|
||||
)
|
||||
|
||||
if rows != 1:
|
||||
raise DatabaseException('Could not update the status "{status}" for checkbox {name} ({id})'.format( # noqa: E501
|
||||
status=name,
|
||||
name=self.fields.name,
|
||||
id=self.fields.id,
|
||||
))
|
||||
|
@ -1,3 +1,3 @@
|
||||
UPDATE "bricktracker_set_checkboxes"
|
||||
SET "name" = :safe_name
|
||||
SET "{{field}}" = :value
|
||||
WHERE "bricktracker_set_checkboxes"."id" IS NOT DISTINCT FROM :id
|
@ -1,3 +0,0 @@
|
||||
UPDATE "bricktracker_set_checkboxes"
|
||||
SET "{{name}}" = :status
|
||||
WHERE "bricktracker_set_checkboxes"."id" IS NOT DISTINCT FROM :id
|
@ -2,9 +2,9 @@ INSERT INTO "bricktracker_set_statuses" (
|
||||
"id",
|
||||
"{{name}}"
|
||||
) VALUES (
|
||||
:id,
|
||||
:status
|
||||
:set_id,
|
||||
:state
|
||||
)
|
||||
ON CONFLICT("id")
|
||||
DO UPDATE SET "{{name}}" = :status
|
||||
WHERE "bricktracker_set_statuses"."id" IS NOT DISTINCT FROM :id
|
||||
DO UPDATE SET "{{name}}" = :state
|
||||
WHERE "bricktracker_set_statuses"."id" IS NOT DISTINCT FROM :set_id
|
||||
|
@ -1,5 +1,3 @@
|
||||
import logging
|
||||
|
||||
from flask import (
|
||||
Blueprint,
|
||||
jsonify,
|
||||
@ -15,8 +13,6 @@ from ..exceptions import exception_handler
|
||||
from ...reload import reload
|
||||
from ...set_checkbox import BrickSetCheckbox
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
admin_checkbox_page = Blueprint(
|
||||
'admin_checkbox',
|
||||
__name__,
|
||||
@ -71,23 +67,13 @@ def do_delete(*, id: str) -> Response:
|
||||
return redirect(url_for('admin.admin', open_checkbox=True))
|
||||
|
||||
|
||||
# Change the status of a checkbox
|
||||
@admin_checkbox_page.route('/<id>/status/<name>', methods=['POST'])
|
||||
# Change the field of a checkbox
|
||||
@admin_checkbox_page.route('/<id>/field/<name>', methods=['POST'])
|
||||
@login_required
|
||||
@exception_handler(__file__, json=True)
|
||||
def update_status(*, id: str, name: str) -> Response:
|
||||
value: bool = request.json.get('value', False) # type: ignore
|
||||
|
||||
def update_field(*, id: str, name: str) -> Response:
|
||||
checkbox = BrickSetCheckbox().select_specific(id)
|
||||
checkbox.update_status(name, value)
|
||||
|
||||
# Info
|
||||
logger.info('Checkbox {name} ({id}): status "{status}" changed to "{state}"'.format( # noqa: E501
|
||||
name=checkbox.fields.name,
|
||||
id=checkbox.fields.id,
|
||||
status=name,
|
||||
state=value,
|
||||
))
|
||||
value = checkbox.update_field(name, json=request.json)
|
||||
|
||||
reload()
|
||||
|
||||
|
@ -37,26 +37,16 @@ def list() -> str:
|
||||
|
||||
|
||||
# Change the status of a checkbox
|
||||
@set_page.route('/<id>/status/<checkbox_id>', methods=['POST'])
|
||||
@set_page.route('/<id>/status/<metadata_id>', methods=['POST'])
|
||||
@login_required
|
||||
@exception_handler(__file__, json=True)
|
||||
def update_status(*, id: str, checkbox_id: str) -> Response:
|
||||
value: bool = request.json.get('value', False) # type: ignore
|
||||
|
||||
def update_status(*, id: str, metadata_id: str) -> Response:
|
||||
brickset = BrickSet().select_light(id)
|
||||
checkbox = BrickSetCheckboxList().get(checkbox_id)
|
||||
checkbox = BrickSetCheckboxList().get(metadata_id)
|
||||
|
||||
brickset.update_status(checkbox, value)
|
||||
state = checkbox.update_set_state(brickset, request.json)
|
||||
|
||||
# Info
|
||||
logger.info('Set {set} ({id}): status "{status}" changed to "{state}"'.format( # noqa: E501
|
||||
set=brickset.fields.set,
|
||||
id=brickset.fields.id,
|
||||
status=checkbox.fields.name,
|
||||
state=value,
|
||||
))
|
||||
|
||||
return jsonify({'value': value})
|
||||
return jsonify({'value': state})
|
||||
|
||||
|
||||
# Ask for deletion of a set
|
||||
|
@ -18,7 +18,7 @@
|
||||
<div class="col-12">
|
||||
<div class="form-check">
|
||||
<input class="form-check-input" type="checkbox" id="grid-{{ checkbox.fields.id }}"
|
||||
data-changer-id="{{ checkbox.fields.id }}" data-changer-prefix="grid" data-changer-url="{{ url_for('admin_checkbox.update_status', id=checkbox.fields.id, name='displayed_on_grid')}}"
|
||||
data-changer-id="{{ checkbox.fields.id }}" data-changer-prefix="grid" data-changer-url="{{ url_for('admin_checkbox.update_field', id=checkbox.fields.id, name='displayed_on_grid')}}"
|
||||
{% if checkbox.fields.displayed_on_grid %}checked{% endif %} autocomplete="off">
|
||||
<label class="form-check-label" for="grid-{{ checkbox.fields.id }}">
|
||||
<i class="ri-grid-line"></i> Displayed on the Set Grid
|
||||
|
@ -30,7 +30,7 @@
|
||||
<ul class="list-group list-group-flush card-check border-bottom-0">
|
||||
{% for checkbox in brickset_checkboxes %}
|
||||
<li class="list-group-item {% if not solo %}p-1{% endif %}">
|
||||
{{ form.checkbox(checkbox.as_dataset(), item.fields.id, checkbox.fields.name, checkbox.status_url(item.fields.id), item.fields[checkbox.as_column()], delete=delete) }}
|
||||
{{ form.checkbox(checkbox.as_dataset(), item.fields.id, checkbox.fields.name, checkbox.url_for_set_state(item.fields.id), item.fields[checkbox.as_column()], delete=delete) }}
|
||||
</li>
|
||||
{% endfor %}
|
||||
</ul>
|
||||
|
Loading…
Reference in New Issue
Block a user