ffdd3d13-9fd2-42d6-8817-f9a076d0bad6

Committed 21 Jan 2026 02:47AM UTC coverage: 79.551% (-8.3%) from 87.869%

Build # ffdd3d13-9fd2-42d6-8817-f9a076d0bad6

Build Type

Pull #239

circleci

Committed by

ThrawnCA

Commit Message

[QOLSVC-12515] add support for specifying facet sorts via query parameters

- Use '_{facet}_sort' as a parallel to '_{facet}_limit', with comma separation for the ordering, eg
'_tags_sort=count,desc' to mimic the default popularity-based sorting.

Pull Request Pull Request #239: QOLSVC-12515 alphabetical facet sort

Run Details

11 of 35 new or added lines in 2 files covered. (31.43%)

13656 existing lines in 309 files now uncovered.

42766 of 53759 relevant lines covered (79.55%)

1.63 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

74.6

/ckanext/example_idatastorebackend/example_sqlite.py

# type: ignore
from __future__ import annotations

import logging
from typing import Any
import sqlalchemy as sa

from ckanext.datastore.backend import DatastoreBackend

log = logging.getLogger(__name__)


class DatastoreExampleSqliteBackend(DatastoreBackend):

    def __init__(self):
        self._engine = None

    def execute(self, sql: str, params: dict[str, Any] | None = None):
        with self._get_engine().begin() as conn:
            return conn.execute(sa.text(sql), params)

    def _get_engine(self):
        if not self._engine:
            self._engine = sa.create_engine(self.write_url)
        return self._engine

    def _insert_records(self, table, records):
        if len(records):
            for record in records:
                sql = sa.insert(
                    sa.table(table, *map(sa.column, record))
                ).values(record)
                self.execute(sql)

    def configure(self, config):
        self.write_url = config.get(
            u'ckan.datastore.write_url'
        ).replace(u'example-', u'')

        return config

    def create(self, context, data_dict, plugin_data):
        columns = str(u', '.join(
            [str(sa.column(e['id'])) + " text" for e in data_dict['fields']]))

        self.execute(sa.text(
            'CREATE TABLE IF NOT EXISTS "{name}"({columns});'.format(
                name=sa.table(data_dict['resource_id']),
                columns=columns
            )
        ))
        self._insert_records(data_dict['resource_id'], data_dict['records'])
        return data_dict

    def upsert(self, context, data_dict):
        raise NotImplementedError()

    def delete(self, context, data_dict):
        self.execute('DROP TABLE IF EXISTS "{0}"'.format(
            data_dict['resource_id']
        ))
        return data_dict

    def search(self, context, data_dict):
        result = self.execute('SELECT * FROM "{0}" LIMIT {1}'.format(
            data_dict['resource_id'],
            data_dict.get(u'limit', 10)
        ))

        data_dict['records'] = list(map(dict, result.fetchall()))
        data_dict['total'] = len(data_dict['records'])

        fields_info = []
        for name, type in self.resource_fields(
                data_dict['resource_id'])['schema'].items():
            fields_info.append({
                u'type': type,
                u'id': name
            })
        data_dict['fields'] = fields_info
        return data_dict

    def search_sql(self, context, data_dict):
        raise NotImplementedError()

    def make_private(self, context, data_dict):
        pass

    def make_public(self, context, data_dict):
        pass

    def resource_exists(self, id):
        return self.execute(
            'select name from sqlite_master where ' +
            f'type = "table" and name = "{id}"'
        ).fetchone()

    def resource_fields(self, id):
        info = self.execute(
            'PRAGMA table_info("{0}")'.format(id)
        ).fetchall()

        schema = {}
        for col in info:
            schema[col.name] = col.type
        return {u'schema': schema, u'meta': {}}

    def resource_id_from_alias(self, alias):
        if self.resource_exists(alias):
            return True, alias
        return False, alias

    def get_all_ids(self):
        return [t.name for t in self.execute(
            '''select name from sqlite_master
            where type = "table"'''
        ).fetchall()]

1	# type: ignore
2	from __future__ import annotations	3✔
3
4	import logging	3✔
5	from typing import Any	3✔
6	import sqlalchemy as sa	3✔
7
8	from ckanext.datastore.backend import DatastoreBackend	3✔
9
10	log = logging.getLogger(__name__)	3✔
11
12
13	class DatastoreExampleSqliteBackend(DatastoreBackend):	3✔
14
15	def __init__(self):	3✔
UNCOV 16	self._engine = None	2✔
17
18	def execute(self, sql: str, params: dict[str, Any] \| None = None):	3✔
19	with self._get_engine().begin() as conn:	×
20	return conn.execute(sa.text(sql), params)	×
21
22	def _get_engine(self):	3✔
23	if not self._engine:	×
24	self._engine = sa.create_engine(self.write_url)	×
25	return self._engine	×
26
27	def _insert_records(self, table, records):	3✔
UNCOV 28	if len(records):	1✔
UNCOV 29	for record in records:	1✔
UNCOV 30	sql = sa.insert(	1✔
31	sa.table(table, *map(sa.column, record))
32	).values(record)
UNCOV 33	self.execute(sql)	1✔
34
35	def configure(self, config):	3✔
UNCOV 36	self.write_url = config.get(	2✔
37	u'ckan.datastore.write_url'
38	).replace(u'example-', u'')
39
UNCOV 40	return config	2✔
41
42	def create(self, context, data_dict, plugin_data):	3✔
UNCOV 43	columns = str(u', '.join(	1✔
44	[str(sa.column(e['id'])) + " text" for e in data_dict['fields']]))
45
UNCOV 46	self.execute(sa.text(	1✔
47	'CREATE TABLE IF NOT EXISTS "{name}"({columns});'.format(
48	name=sa.table(data_dict['resource_id']),
49	columns=columns
50	)
51	))
UNCOV 52	self._insert_records(data_dict['resource_id'], data_dict['records'])	1✔
UNCOV 53	return data_dict	1✔
54
55	def upsert(self, context, data_dict):	3✔
56	raise NotImplementedError()	×
57
58	def delete(self, context, data_dict):	3✔
UNCOV 59	self.execute('DROP TABLE IF EXISTS "{0}"'.format(	1✔
60	data_dict['resource_id']
61	))
UNCOV 62	return data_dict	1✔
63
64	def search(self, context, data_dict):	3✔
UNCOV 65	result = self.execute('SELECT * FROM "{0}" LIMIT {1}'.format(	1✔
66	data_dict['resource_id'],
67	data_dict.get(u'limit', 10)
68	))
69
UNCOV 70	data_dict['records'] = list(map(dict, result.fetchall()))	1✔
UNCOV 71	data_dict['total'] = len(data_dict['records'])	1✔
72
UNCOV 73	fields_info = []	1✔
UNCOV 74	for name, type in self.resource_fields(	1✔
75	data_dict['resource_id'])['schema'].items():
UNCOV 76	fields_info.append({	1✔
77	u'type': type,
78	u'id': name
79	})
UNCOV 80	data_dict['fields'] = fields_info	1✔
UNCOV 81	return data_dict	1✔
82
83	def search_sql(self, context, data_dict):	3✔
84	raise NotImplementedError()	×
85
86	def make_private(self, context, data_dict):	3✔
87	pass	×
88
89	def make_public(self, context, data_dict):	3✔
90	pass	×
91
92	def resource_exists(self, id):	3✔
UNCOV 93	return self.execute(	1✔
94	'select name from sqlite_master where ' +
95	f'type = "table" and name = "{id}"'
96	).fetchone()
97
98	def resource_fields(self, id):	3✔
99	info = self.execute(	×
100	'PRAGMA table_info("{0}")'.format(id)
101	).fetchall()
102
103	schema = {}	×
104	for col in info:	×
105	schema[col.name] = col.type	×
106	return {u'schema': schema, u'meta': {}}	×
107
108	def resource_id_from_alias(self, alias):	3✔
UNCOV 109	if self.resource_exists(alias):	1✔
UNCOV 110	return True, alias	1✔
111	return False, alias	×
112
113	def get_all_ids(self):	3✔
114	return [t.name for t in self.execute(	×
115	'''select name from sqlite_master
116	where type = "table"'''
117	).fetchall()]

qld-gov-au / ckan / ffdd3d13-9fd2-42d6-8817-f9a076d0bad6

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous