openstack/glance
Code Issues Proposed changes

Add user_get_image_count() to DB API

Browse Source 
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
This commit is contained in:
Dan Smith 2021-04-27 06:47:58 -07:00
parent 59990d513a
commit d3d6a646e3
2 changed files with 19 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
glance/db/sqlalchemy/api.py
View File

@ -781,6 +781,13 @@ def _image_get_staging_usage_by_owner(owner, session):
copying_images)) 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): def _validate_image(values, mandatory_status=True):
""" """
Validates the incoming data and raises a Invalid exception 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) 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): def _task_info_format(task_info_ref):
"""Format a task info ref for consumption outside of this module""" """Format a task info ref for consumption outside of this module"""
if task_info_ref is None: if task_info_ref is None:

7
glance/tests/functional/db/test_sqlalchemy.py
View File

@ -418,3 +418,10 @@ class TestImageStorageUsage(base.TestDriver,
# Each user has two active images of size 100 each, but only one # Each user has two active images of size 100 each, but only one
# has an active location. # has an active location.
self.assertEqual(100, usage) 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)