wutta/wuttaweb
3
0
Fork 0
Code Issues Activity

fix: fix 'duplicate-code' for pylint

Browse source
This commit is contained in:
Lance Edgar 2025-09-01 13:31:33 -05:00
parent ad74bede04
commit a1868e1f44
10 changed files with 233 additions and 129 deletions
Download patch file Download diff file Expand all files Collapse all files

5
.pylintrc
View file

@ -2,4 +2,7 @@
[MESSAGES CONTROL]
disable=fixme,
duplicate-code,
[SIMILARITIES]
# nb. cuts out some noise for duplicate-code
min-similarity-lines=5

15
src/wuttaweb/forms/base.py
View file

@ -37,7 +37,13 @@ from colanderalchemy import SQLAlchemySchemaNode
from pyramid.renderers import render
from webhelpers2.html import HTML
from wuttaweb.util import FieldList, get_form_data, get_model_fields, make_json_safe
from wuttaweb.util import (
FieldList,
get_form_data,
get_model_fields,
make_json_safe,
render_vue_finalize,
)
log = logging.getLogger(__name__)
@ -1095,12 +1101,7 @@ class Form: # pylint: disable=too-many-instance-attributes,too-many-public-meth
The actual output may depend on various form attributes, in
particular :attr:`vue_tagname`.
"""
set_data = f"{self.vue_component}.data = function() {{ return {self.vue_component}Data }}"
make_component = f"Vue.component('{self.vue_tagname}', {self.vue_component})"
return HTML.tag(
"script",
c=["\n", HTML.literal(set_data), "\n", HTML.literal(make_component), "\n"],
)
return render_vue_finalize(self.vue_tagname, self.vue_component)
def get_vue_model_data(self):
"""

14
src/wuttaweb/grids/base.py
View file

@ -39,7 +39,12 @@ from pyramid.renderers import render
from webhelpers2.html import HTML
from wuttjamaican.db.util import UUID
from wuttaweb.util import FieldList, get_model_fields, make_json_safe
from wuttaweb.util import (
FieldList,
get_model_fields,
make_json_safe,
render_vue_finalize,
)
from wuttaweb.grids.filters import default_sqlalchemy_filters, VerbNotSupported
@ -2212,12 +2217,7 @@ class Grid: # pylint: disable=too-many-instance-attributes,too-many-public-meth
The actual output may depend on various grid attributes, in
particular :attr:`vue_tagname`.
"""
set_data = f"{self.vue_component}.data = function() {{ return {self.vue_component}Data }}"
make_component = f"Vue.component('{self.vue_tagname}', {self.vue_component})"
return HTML.tag(
"script",
c=["\n", HTML.literal(set_data), "\n", HTML.literal(make_component), "\n"],
)
return render_vue_finalize(self.vue_tagname, self.vue_component)
def get_vue_columns(self):
"""

65
src/wuttaweb/util.py
View file

@ -618,6 +618,71 @@ def make_json_safe(value, key=None, warn=True):
return value
def render_vue_finalize(vue_tagname, vue_component):
"""
Render the Vue "finalize" script for a form or grid component.
This is a convenience for shared logic; it returns e.g.:
.. code-block:: html
<script>
WuttaGrid.data = function() { return WuttaGridData }
Vue.component('wutta-grid', WuttaGrid)
</script>
"""
set_data = f"{vue_component}.data = function() {{ return {vue_component}Data }}"
make_component = f"Vue.component('{vue_tagname}', {vue_component})"
return HTML.tag(
"script",
c=["\n", HTML.literal(set_data), "\n", HTML.literal(make_component), "\n"],
)
def make_users_grid(request, **kwargs):
"""
Make and return a users (sub)grid.
This grid is shown for the Users field when viewing a Person or
Role, for instance. It is called by the following methods:
* :meth:`wuttaweb.views.people.PersonView.make_users_grid()`
* :meth:`wuttaweb.views.roles.RoleView.make_users_grid()`
:returns: Fully configured :class:`~wuttaweb.grids.base.Grid`
instance.
"""
config = request.wutta_config
app = config.get_app()
model = app.model
web = app.get_web_handler()
if "key" not in kwargs:
route_prefix = kwargs.pop("route_prefix")
kwargs["key"] = f"{route_prefix}.view.users"
kwargs.setdefault("model_class", model.User)
grid = web.make_grid(request, **kwargs)
if request.has_perm("users.view"):
def view_url(user, i): # pylint: disable=unused-argument
return request.route_url("users.view", uuid=user.uuid)
grid.add_action("view", icon="eye", url=view_url)
grid.set_link("person")
grid.set_link("username")
if request.has_perm("users.edit"):
def edit_url(user, i): # pylint: disable=unused-argument
return request.route_url("users.edit", uuid=user.uuid)
grid.add_action("edit", url=edit_url)
return grid
##############################
# theme functions
##############################

60
src/wuttaweb/views/batch.py
View file

@ -51,6 +51,8 @@ class BatchMasterView(MasterView):
from :meth:`get_batch_handler()`.
"""
executable = True
labels = {
"id": "Batch ID",
"status_code": "Status",
@ -330,29 +332,22 @@ class BatchMasterView(MasterView):
raise RuntimeError("can't find the batch")
time.sleep(0.1)
try:
# populate the batch
self.batch_handler.do_populate(batch, progress=progress)
session.flush()
except Exception as error: # pylint: disable=broad-exception-caught
session.rollback()
def onerror():
log.warning(
"failed to populate %s: %s",
self.get_model_title(),
batch,
exc_info=True,
)
if progress:
progress.handle_error(error)
else:
session.commit()
if progress:
progress.handle_success()
finally:
session.close()
self.do_thread_body(
self.batch_handler.do_populate,
(batch,),
{"progress": progress},
onerror,
session=session,
progress=progress,
)
##############################
# execute methods
@ -418,36 +413,3 @@ class BatchMasterView(MasterView):
):
""" """
return row.STATUS.get(value, value)
##############################
# configuration
##############################
@classmethod
def defaults(cls, config): # pylint: disable=empty-docstring
""" """
cls._defaults(config)
cls._batch_defaults(config)
@classmethod
def _batch_defaults(cls, config):
route_prefix = cls.get_route_prefix()
permission_prefix = cls.get_permission_prefix()
model_title = cls.get_model_title()
instance_url_prefix = cls.get_instance_url_prefix()
# execute
config.add_route(
f"{route_prefix}.execute",
f"{instance_url_prefix}/execute",
request_method="POST",
)
config.add_view(
cls,
attr="execute",
route_name=f"{route_prefix}.execute",
permission=f"{permission_prefix}.execute",
)
config.add_wutta_permission(
permission_prefix, f"{permission_prefix}.execute", f"Execute {model_title}"
)

82
src/wuttaweb/views/master.py
View file

@ -822,33 +822,25 @@ class MasterView(View): # pylint: disable=too-many-public-methods
self, query, progress=None
):
""" """
model_title_plural = self.get_model_title_plural()
# nb. use new session, separate from web transaction
session = self.app.make_session()
records = query.with_session(session).all()
try:
self.delete_bulk_action(records, progress=progress)
except Exception as error: # pylint: disable=broad-exception-caught
session.rollback()
def onerror():
log.warning(
"failed to delete %s results for %s",
len(records),
model_title_plural,
self.get_model_title_plural(),
exc_info=True,
)
if progress:
progress.handle_error(error)
else:
session.commit()
if progress:
progress.handle_success()
finally:
session.close()
self.do_thread_body(
self.delete_bulk_action,
(records,),
{"progress": progress},
onerror,
session=session,
progress=progress,
)
def delete_bulk_action(self, data, progress=None):
"""
@ -2485,6 +2477,60 @@ class MasterView(View): # pylint: disable=too-many-public-methods
session = session or self.Session()
session.add(obj)
def do_thread_body( # pylint: disable=too-many-arguments,too-many-positional-arguments
self, func, args, kwargs, onerror=None, session=None, progress=None
):
"""
Generic method to invoke for thread operations.
:param func: Callable which performs the actual logic. This
will be wrapped with a try/except statement for error
handling.
:param args: Tuple of positional arguments to pass to the
``func`` callable.
:param kwargs: Dict of keyword arguments to pass to the
``func`` callable.
:param onerror: Optional callback to invoke if ``func`` raises
an error. It should not expect any arguments.
:param session: Optional :term:`db session` in effect. Note
that if supplied, it will be *committed* (or rolled back on
error) and *closed* by this method. If you need more
specialized handling, do not use this method (or don't
specify the ``session``).
:param progress: Optional progress factory. If supplied, this
is assumed to be a
:class:`~wuttaweb.progress.SessionProgress` instance, and
it will be updated per success or failure of ``func``
invocation.
"""
try:
func(*args, **kwargs)
except Exception as error: # pylint: disable=broad-exception-caught
if session:
session.rollback()
if onerror:
onerror()
else:
log.warning("failed to invoke thread callable: %s", func, exc_info=True)
if progress:
progress.handle_error(error)
else:
if session:
session.commit()
if progress:
progress.handle_success()
finally:
if session:
session.close()
##############################
# row methods
##############################

27
src/wuttaweb/views/people.py
View file

@ -28,6 +28,7 @@ import sqlalchemy as sa
from wuttjamaican.db.model import Person
from wuttaweb.views import MasterView
from wuttaweb.util import make_users_grid
class PersonView(MasterView): # pylint: disable=abstract-method
@ -107,12 +108,9 @@ class PersonView(MasterView): # pylint: disable=abstract-method
:returns: Fully configured :class:`~wuttaweb.grids.base.Grid`
instance.
"""
model = self.app.model
route_prefix = self.get_route_prefix()
grid = self.make_grid(
key=f"{route_prefix}.view.users",
model_class=model.User,
return make_users_grid(
self.request,
route_prefix=self.get_route_prefix(),
data=person.users,
columns=[
"username",
@ -120,23 +118,6 @@ class PersonView(MasterView): # pylint: disable=abstract-method
],
)
if self.request.has_perm("users.view"):
def view_url(user, i): # pylint: disable=unused-argument
return self.request.route_url("users.view", uuid=user.uuid)
grid.add_action("view", icon="eye", url=view_url)
grid.set_link("username")
if self.request.has_perm("users.edit"):
def edit_url(user, i): # pylint: disable=unused-argument
return self.request.route_url("users.edit", uuid=user.uuid)
grid.add_action("edit", url=edit_url)
return grid
def objectify(self, form): # pylint: disable=empty-docstring
""" """
person = super().objectify(form)

28
src/wuttaweb/views/roles.py
View file

@ -29,6 +29,7 @@ from wuttaweb.views import MasterView
from wuttaweb.db import Session
from wuttaweb.forms import widgets
from wuttaweb.forms.schema import Permissions, RoleRef
from wuttaweb.util import make_users_grid
class RoleView(MasterView): # pylint: disable=abstract-method
@ -151,12 +152,9 @@ class RoleView(MasterView): # pylint: disable=abstract-method
:returns: Fully configured :class:`~wuttaweb.grids.base.Grid`
instance.
"""
model = self.app.model
route_prefix = self.get_route_prefix()
grid = self.make_grid(
key=f"{route_prefix}.view.users",
model_class=model.User,
return make_users_grid(
self.request,
route_prefix=self.get_route_prefix(),
data=role.users,
columns=[
"username",
@ -165,24 +163,6 @@ class RoleView(MasterView): # pylint: disable=abstract-method
],
)
if self.request.has_perm("users.view"):
def view_url(user, i): # pylint: disable=unused-argument
return self.request.route_url("users.view", uuid=user.uuid)
grid.add_action("view", icon="eye", url=view_url)
grid.set_link("person")
grid.set_link("username")
if self.request.has_perm("users.edit"):
def edit_url(user, i): # pylint: disable=unused-argument
return self.request.route_url("users.edit", uuid=user.uuid)
grid.add_action("edit", url=edit_url)
return grid
def unique_name(self, node, value): # pylint: disable=empty-docstring
""" """
model = self.app.model

47
tests/test_util.py
View file

@ -16,6 +16,7 @@ from wuttjamaican.util import resource_path
from wuttaweb import util as mod
from wuttaweb.app import establish_theme
from wuttaweb.grids import Grid
from wuttaweb.testing import WebTestCase
@ -665,6 +666,52 @@ class TestMakeJsonSafe(TestCase):
)
class TestRenderVueFinalize(TestCase):
def basic(self):
html = mod.render_vue_finalize("wutta-grid", "WuttaGrid")
self.assertIn("<script>", html)
self.assertIn("Vue.component('wutta-grid', WuttaGrid)", html)
class TestMakeUsersGrid(WebTestCase):
def test_make_users_grid(self):
self.pyramid_config.add_route("users.view", "/users/{uuid}/view")
self.pyramid_config.add_route("users.edit", "/users/{uuid}/edit")
model = self.app.model
person = model.Person(full_name="John Doe")
self.session.add(person)
user = model.User(username="john", person=person)
self.session.add(user)
self.session.commit()
# basic (no actions because not prvileged)
grid = mod.make_users_grid(self.request, key="blah.users", data=person.users)
self.assertIsInstance(grid, Grid)
self.assertFalse(grid.linked_columns)
self.assertFalse(grid.actions)
# key may be derived from route_prefix
grid = mod.make_users_grid(self.request, route_prefix="foo")
self.assertIsInstance(grid, Grid)
self.assertEqual(grid.key, "foo.view.users")
# view + edit actions (because root)
with patch.object(self.request, "is_root", new=True):
grid = mod.make_users_grid(
self.request, key="blah.users", data=person.users
)
self.assertIsInstance(grid, Grid)
self.assertIn("username", grid.linked_columns)
self.assertEqual(len(grid.actions), 2)
self.assertEqual(grid.actions[0].key, "view")
self.assertEqual(grid.actions[1].key, "edit")
# render grid to ensure coverage for link urls
grid.render_vue_template()
class TestGetAvailableThemes(TestCase):
def setUp(self):

19
tests/views/test_master.py
View file

@ -1371,6 +1371,25 @@ class TestMasterView(WebTestCase):
self.session.commit()
self.assertEqual(self.session.query(model.Setting).count(), 7)
def test_do_thread_body(self):
view = self.make_view()
# nb. so far this is just proving coverage, in case caller
# does not specify an error handler
def func():
raise RuntimeError
# with error handler
onerror = MagicMock()
view.do_thread_body(func, (), {}, onerror)
onerror.assert_called_once_with()
# without error handler
onerror.reset_mock()
view.do_thread_body(func, (), {})
onerror.assert_not_called()
def test_delete_bulk_thread(self):
self.pyramid_config.add_route("settings", "/settings/")
model = self.app.model