#6761

Committed 23 Sep 2025 01:49PM UTC coverage: 78.293% (+3.4%) from 74.853%

Build # #6761

Build Type

Pull #6752

coveralls-python

Committed by

web-flow

Commit Message

Merge pull request #6751 from OCHA-DAP/feature/HDX-11115-coverage-increase-hdx-package

HDX-11115 - hdx_package

Pull Request Pull Request #6752: Feature/hdx 11115 coverage increase

Run Details

68 of 83 new or added lines in 12 files covered. (81.93%)

5 existing lines in 4 files now uncovered.

13464 of 17197 relevant lines covered (78.29%)

0.78 hits per line

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

85.19

/ckanext-hdx_pages/ckanext/hdx_pages/model.py

from __future__ import print_function

import sqlalchemy.orm as orm
import sqlalchemy.types as types
import logging
import datetime
import ckan.model as model
from ckan.model.domain_object import DomainObject
from ckan.model import meta
import ckan.model.types as _types

from sqlalchemy.schema import Table, Column, ForeignKey
from sqlalchemy.engine.reflection import Inspector

mapper = orm.mapper
log = logging.getLogger(__name__)

page_table = None
page_group_association_table = None
page_tag_association_table = None


def setup():
    if page_table is None:
        define_page_table()
        log.debug('Page table defined in memory')

    if page_group_association_table is None:
        define_page_group_association_table()
        log.debug('Page group association table defined in memory')

    if page_tag_association_table is None:
        define_page_tag_association_table()
        log.debug('Page tag association table defined in memory')

    # checks for existence first
    create_table()


class PageBaseModel(DomainObject):
    @classmethod
    def filter(cls, **kwargs):
        return meta.Session.query(cls).filter_by(**kwargs)

    @classmethod
    def get(cls, **kwargs):
        instance = cls.filter(**kwargs).first()
        return instance

    @classmethod
    def exists(cls, **kwargs):
        if cls.filter(**kwargs).first():
            return True
        else:
            return False

    @classmethod
    def create(cls, **kwargs):
        defer_commit = kwargs.get('defer_commit')
        if defer_commit:
            del kwargs['defer_commit']
        instance = cls(**kwargs)
        meta.Session.add(instance)
        if not defer_commit:
            meta.Session.commit()
        return instance.as_dict()


class Page(PageBaseModel):
    """
    Page table
    """

    def __init__(self, name, title, description, type, state, sections, status, extras, modified=None):
        self.name = name
        self.title = title
        self.description = description
        self.type = type
        self.state = state
        self.sections = sections
        self.status = status
        self.extras = extras
        self.modified = modified

    @classmethod
    def get_by_id(cls, id):
        """Returns a group object referenced by its id or name."""
        query = meta.Session.query(cls).filter(cls.id == id)
        page = query.first()
        if page is None:
            page = cls.by_name(id)
        return page

    @classmethod
    def get_by_name(cls, name):
        query = meta.Session.query(cls)
        return query.filter_by(name=name).first()

    @classmethod
    def check_exists(cls):
        return page_table.exists()


def define_page_table():
    global page_table

    page_table = Table('page', meta.metadata,
                       Column('id', types.UnicodeText, primary_key=True, default=_types.make_uuid),
                       Column('name', types.UnicodeText, nullable=False, unique=True, index=True),
                       Column('title', types.UnicodeText, nullable=False),
                       Column('description', types.UnicodeText, nullable=True),
                       Column('type', types.UnicodeText),
                       Column('state', types.UnicodeText),
                       Column('sections', types.UnicodeText),
                       Column('extras', types.UnicodeText),
                       Column('modified', types.DateTime, default=datetime.datetime.now, nullable=False),
                       Column('status', types.UnicodeText),
                       )

    mapper(Page, page_table)


class PageGroupAssociation(PageBaseModel):
    @classmethod
    def get_group_ids_for_page(cls, page_id):
        """
        Return a list of group ids associated with the passed page_id.
        """
        page = Page.get_by_id(page_id)
        # associated_group_id_list = meta.Session.query(cls.group_id).filter_by(page_id=page_id).all()

        result = [assoc.group_id for assoc in page.countries_assoc_all]
        return result


class PageTagAssociation(PageBaseModel):
    @classmethod
    def get_tag_ids_for_page(cls, page_id):
        """
        Return a list of tag ids associated with the passed page_id.
        """
        page = Page.get_by_id(page_id)

        result = [assoc.tag_id for assoc in page.tags_assoc_all]
        return result

    @classmethod
    def get_page_ids_for_tag(cls, tag_id):
        """
        Return a list of page ids associated with the passed tag_id.
        """

        page_tag_list = meta.Session.query(cls.page_id).filter_by(tag_id=tag_id).all()
        result = [res.page_id for res in page_tag_list]
        return result


def define_page_group_association_table():
    global page_group_association_table

    page_group_association_table = Table(
        'page_group_association',
        meta.metadata,
        Column('group_id', types.UnicodeText,
               ForeignKey('group.id', ondelete='CASCADE', onupdate='CASCADE'), primary_key=True, nullable=False),
        Column('page_id', types.UnicodeText,
               ForeignKey('page.id', ondelete='CASCADE', onupdate='CASCADE'), primary_key=True, nullable=False)
    )

    meta.registry.map_imperatively(PageGroupAssociation, page_group_association_table, properties={
        'page': orm.relationship(Page, backref=orm.backref('countries_assoc_all', cascade='all, delete-orphan'))})


def define_page_tag_association_table():
    global page_tag_association_table

    page_tag_association_table = Table(
        'page_tag_association',
        meta.metadata,
        Column('tag_id', types.UnicodeText,
               ForeignKey('tag.id', ondelete='CASCADE', onupdate='CASCADE'), primary_key=True, nullable=False),
        Column('page_id', types.UnicodeText,
               ForeignKey('page.id', ondelete='CASCADE', onupdate='CASCADE'), primary_key=True, nullable=False)
    )

    meta.registry.map_imperatively(PageTagAssociation, page_tag_association_table, properties={
        'page': orm.relationship(Page, backref=orm.backref('tags_assoc_all', cascade='all, delete-orphan'))})


def create_table():
    engine = model.ensure_engine()
    if model.group_table.exists(engine):
        if not page_table.exists(engine):
            page_table.create(engine)
            print('Page table created')
        else:
            patch_table_add_column('extras')

        if not page_group_association_table.exists(engine):
            page_group_association_table.create(engine)
            print('page group association table created')
        if not page_tag_association_table.exists(engine):
            page_tag_association_table.create(engine)
            print('page tag association table created')

#
def delete_table():
    if page_table.exists():
        page_table.delete()
        log.debug('Page table deleted')

def drop_table():
    """
    Drop page table
    """
    if page_table.exists():
        page_table.drop()
        log.debug('Page table dropped')


def patch_table_add_column(column_name):
    table_name = 'page'
    try:
        # print('Starting to patch table %s' % table_name)
        engine = model.meta.engine
        inspector = Inspector.from_engine(engine)
        columns = inspector.get_columns(table_name)

        if not any(column['name'] == column_name for column in columns):
            column = Column(column_name, types.UnicodeText, default='')
            column_name = column.compile(dialect=engine.dialect)
            column_type = column.type.compile(engine.dialect)
            engine.execute(
                'ALTER TABLE %s ADD COLUMN %s %s' % (table_name, column_name, column_type))

        # print('Finish to patch table %s' % table_name)
    except Exception:
        print('There was an error during patching %s table. Column: %s' % (table_name, column_name))

1	from __future__ import print_function	1✔
2
3	import sqlalchemy.orm as orm	1✔
4	import sqlalchemy.types as types	1✔
5	import logging	1✔
6	import datetime	1✔
7	import ckan.model as model	1✔
8	from ckan.model.domain_object import DomainObject	1✔
9	from ckan.model import meta	1✔
10	import ckan.model.types as _types	1✔
11
12	from sqlalchemy.schema import Table, Column, ForeignKey	1✔
13	from sqlalchemy.engine.reflection import Inspector	1✔
14
15	mapper = orm.mapper	1✔
16	log = logging.getLogger(__name__)	1✔
17
18	page_table = None	1✔
19	page_group_association_table = None	1✔
20	page_tag_association_table = None	1✔
21
22
23	def setup():	1✔
24	if page_table is None:	1✔
25	define_page_table()	1✔
26	log.debug('Page table defined in memory')	1✔
27
28	if page_group_association_table is None:	1✔
29	define_page_group_association_table()	1✔
30	log.debug('Page group association table defined in memory')	1✔
31
32	if page_tag_association_table is None:	1✔
33	define_page_tag_association_table()	1✔
34	log.debug('Page tag association table defined in memory')	1✔
35
36	# checks for existence first
37	create_table()	1✔
38
39
40	class PageBaseModel(DomainObject):	1✔
41	@classmethod	1✔
42	def filter(cls, **kwargs):	1✔
43	return meta.Session.query(cls).filter_by(**kwargs)	1✔
44
45	@classmethod	1✔
46	def get(cls, **kwargs):	1✔
47	instance = cls.filter(**kwargs).first()	1✔
48	return instance	1✔
49
50	@classmethod	1✔
51	def exists(cls, **kwargs):	1✔
52	if cls.filter(**kwargs).first():	1✔
53	return True	×
54	else:
55	return False	1✔
56
57	@classmethod	1✔
58	def create(cls, **kwargs):	1✔
59	defer_commit = kwargs.get('defer_commit')	1✔
60	if defer_commit:	1✔
61	del kwargs['defer_commit']	1✔
62	instance = cls(**kwargs)	1✔
63	meta.Session.add(instance)	1✔
64	if not defer_commit:	1✔
65	meta.Session.commit()	×
66	return instance.as_dict()	1✔
67
68
69	class Page(PageBaseModel):	1✔
70	"""
71	Page table
72	"""
73
74	def __init__(self, name, title, description, type, state, sections, status, extras, modified=None):	1✔
75	self.name = name	1✔
76	self.title = title	1✔
77	self.description = description	1✔
78	self.type = type	1✔
79	self.state = state	1✔
80	self.sections = sections	1✔
81	self.status = status	1✔
82	self.extras = extras	1✔
83	self.modified = modified	1✔
84
85	@classmethod	1✔
86	def get_by_id(cls, id):	1✔
87	"""Returns a group object referenced by its id or name."""
88	query = meta.Session.query(cls).filter(cls.id == id)	1✔
89	page = query.first()	1✔
90	if page is None:	1✔
91	page = cls.by_name(id)	1✔
92	return page	1✔
93
94	@classmethod	1✔
95	def get_by_name(cls, name):	1✔
96	query = meta.Session.query(cls)	×
97	return query.filter_by(name=name).first()	×
98
99	@classmethod	1✔
100	def check_exists(cls):	1✔
101	return page_table.exists()	×
102
103
104	def define_page_table():	1✔
105	global page_table
106
107	page_table = Table('page', meta.metadata,	1✔
108	Column('id', types.UnicodeText, primary_key=True, default=_types.make_uuid),
109	Column('name', types.UnicodeText, nullable=False, unique=True, index=True),
110	Column('title', types.UnicodeText, nullable=False),
111	Column('description', types.UnicodeText, nullable=True),
112	Column('type', types.UnicodeText),
113	Column('state', types.UnicodeText),
114	Column('sections', types.UnicodeText),
115	Column('extras', types.UnicodeText),
116	Column('modified', types.DateTime, default=datetime.datetime.now, nullable=False),
117	Column('status', types.UnicodeText),
118	)
119
120	mapper(Page, page_table)	1✔
121
122
123	class PageGroupAssociation(PageBaseModel):	1✔
124	@classmethod	1✔
125	def get_group_ids_for_page(cls, page_id):	1✔
126	"""
127	Return a list of group ids associated with the passed page_id.
128	"""
129	page = Page.get_by_id(page_id)	1✔
130	# associated_group_id_list = meta.Session.query(cls.group_id).filter_by(page_id=page_id).all()
131
132	result = [assoc.group_id for assoc in page.countries_assoc_all]	1✔
133	return result	1✔
134
135
136	class PageTagAssociation(PageBaseModel):	1✔
137	@classmethod	1✔
138	def get_tag_ids_for_page(cls, page_id):	1✔
139	"""
140	Return a list of tag ids associated with the passed page_id.
141	"""
142	page = Page.get_by_id(page_id)	1✔
143
144	result = [assoc.tag_id for assoc in page.tags_assoc_all]	1✔
145	return result	1✔
146
147	@classmethod	1✔
148	def get_page_ids_for_tag(cls, tag_id):	1✔
149	"""
150	Return a list of page ids associated with the passed tag_id.
151	"""
152
153	page_tag_list = meta.Session.query(cls.page_id).filter_by(tag_id=tag_id).all()	×
154	result = [res.page_id for res in page_tag_list]	×
155	return result	×
156
157
158	def define_page_group_association_table():	1✔
159	global page_group_association_table
160
161	page_group_association_table = Table(	1✔
162	'page_group_association',
163	meta.metadata,
164	Column('group_id', types.UnicodeText,
165	ForeignKey('group.id', ondelete='CASCADE', onupdate='CASCADE'), primary_key=True, nullable=False),
166	Column('page_id', types.UnicodeText,
167	ForeignKey('page.id', ondelete='CASCADE', onupdate='CASCADE'), primary_key=True, nullable=False)
168	)
169
170	meta.registry.map_imperatively(PageGroupAssociation, page_group_association_table, properties={	1✔
171	'page': orm.relationship(Page, backref=orm.backref('countries_assoc_all', cascade='all, delete-orphan'))})
172
173
174	def define_page_tag_association_table():	1✔
175	global page_tag_association_table
176
177	page_tag_association_table = Table(	1✔
178	'page_tag_association',
179	meta.metadata,
180	Column('tag_id', types.UnicodeText,
181	ForeignKey('tag.id', ondelete='CASCADE', onupdate='CASCADE'), primary_key=True, nullable=False),
182	Column('page_id', types.UnicodeText,
183	ForeignKey('page.id', ondelete='CASCADE', onupdate='CASCADE'), primary_key=True, nullable=False)
184	)
185
186	meta.registry.map_imperatively(PageTagAssociation, page_tag_association_table, properties={	1✔
187	'page': orm.relationship(Page, backref=orm.backref('tags_assoc_all', cascade='all, delete-orphan'))})
188
189
190	def create_table():	1✔
191	engine = model.ensure_engine()	1✔
192	if model.group_table.exists(engine):	1✔
193	if not page_table.exists(engine):	1✔
194	page_table.create(engine)	1✔
195	print('Page table created')	1✔
196	else:
197	patch_table_add_column('extras')	1✔
198
199	if not page_group_association_table.exists(engine):	1✔
200	page_group_association_table.create(engine)	1✔
201	print('page group association table created')	1✔
202	if not page_tag_association_table.exists(engine):	1✔
203	page_tag_association_table.create(engine)	1✔
204	print('page tag association table created')	1✔
205
206	#
207	def delete_table():	1✔
NEW 208	if page_table.exists():	×
NEW 209	page_table.delete()	×
NEW 210	log.debug('Page table deleted')	×
211
212	def drop_table():	1✔
213	"""
214	Drop page table
215	"""
NEW 216	if page_table.exists():	×
NEW 217	page_table.drop()	×
NEW 218	log.debug('Page table dropped')	×
219
220
221	def patch_table_add_column(column_name):	1✔
222	table_name = 'page'	1✔
223	try:	1✔
224	# print('Starting to patch table %s' % table_name)
225	engine = model.meta.engine	1✔
226	inspector = Inspector.from_engine(engine)	1✔
227	columns = inspector.get_columns(table_name)	1✔
228
229	if not any(column['name'] == column_name for column in columns):	1✔
230	column = Column(column_name, types.UnicodeText, default='')	×
231	column_name = column.compile(dialect=engine.dialect)	×
232	column_type = column.type.compile(engine.dialect)	×
233	engine.execute(	×
234	'ALTER TABLE %s ADD COLUMN %s %s' % (table_name, column_name, column_type))
235
236	# print('Finish to patch table %s' % table_name)
237	except Exception:	×
238	print('There was an error during patching %s table. Column: %s' % (table_name, column_name))	×

OCHA-DAP / hdx-ckan / #6761

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