From d3d6a646e32ffe8f59e17fa02b7b5ecf06d7a417 Mon Sep 17 00:00:00 2001 From: Dan Smith Date: Tue, 27 Apr 2021 06:47:58 -0700 Subject: [PATCH] Add user_get_image_count() to DB API This adds an efficient query method for getting a count of a user's non-deleted images. It can be used by subsequent patches to enforce the image count quota. Partially-implements: blueprint glance-unified-quotas Change-Id: I4d6e071b8d48862032f36c80ee19632a3fe85009 --- glance/db/sqlalchemy/api.py | 12 ++++++++++++ glance/tests/functional/db/test_sqlalchemy.py | 7 +++++++ 2 files changed, 19 insertions(+) diff --git a/glance/db/sqlalchemy/api.py b/glance/db/sqlalchemy/api.py index 18094a7aab..60dfad33b8 100644 --- a/glance/db/sqlalchemy/api.py +++ b/glance/db/sqlalchemy/api.py @@ -781,6 +781,13 @@ def _image_get_staging_usage_by_owner(owner, session): copying_images)) +def _image_get_count_by_owner(owner, session): + query = session.query(models.Image) + query = query.filter(models.Image.owner == owner) + query = query.filter(~models.Image.status.in_(['killed', 'deleted'])) + return query.count() + + def _validate_image(values, mandatory_status=True): """ Validates the incoming data and raises a Invalid exception @@ -1594,6 +1601,11 @@ def user_get_staging_usage(context, owner_id, session=None): return _image_get_staging_usage_by_owner(owner_id, session) +def user_get_image_count(context, owner_id, session=None): + session = session or get_session() + return _image_get_count_by_owner(owner_id, session) + + def _task_info_format(task_info_ref): """Format a task info ref for consumption outside of this module""" if task_info_ref is None: diff --git a/glance/tests/functional/db/test_sqlalchemy.py b/glance/tests/functional/db/test_sqlalchemy.py index 8877694325..8c9eaae1b2 100644 --- a/glance/tests/functional/db/test_sqlalchemy.py +++ b/glance/tests/functional/db/test_sqlalchemy.py @@ -418,3 +418,10 @@ class TestImageStorageUsage(base.TestDriver, # Each user has two active images of size 100 each, but only one # has an active location. self.assertEqual(100, usage) + + def test_get_image_count(self): + for owner, ctxt in self.contexts.items(): + count = self.db_api.user_get_image_count(ctxt, ctxt.owner) + # Each user has two active images, two staged images, two + # importing, and two queued images + self.assertEqual(8, count)