1481284286

Build Type

github

Committed by GitHub

Commit Message

<a href="https://github.com/adsabs/resolver_service/commit/<a class=hub.com/adsabs/resolver_service/commit/<a class="double-link" href="https://git"><a class=hub.com/adsabs/resolver_service/commit/<a class="double-link" href="https://git"><a class=hub.com/adsabs/resolver_service/commit/<a class="double-link" href="https://git"><a class=hub.com/adsabs/resolver_service/commit/<a class="double-link" href="https://git"><a class=hub.com/adsabs/resolver_service/commit/d636836228d121ccafcfcbfa4163d0c37437c21b">d63683622">&lt;a href=&quot;https://github.com/adsabs/resolver_service/commit/</a><a class="double-link" href="https://github.com/adsabs/resolver_service/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/adsabs/resolver_service/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/adsabs/resolver_service/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/adsabs/resolver_service/commit/d636836228d121ccafcfcbfa4163d0c37437c21b">d63683622</a><a href="https://github.com/adsabs/resolver_service/commit/d636836228d121ccafcfcbfa4163d0c37437c21b">&lt;a href=&quot;https://github.com/adsabs/resolver_service/commit/d636836228d121ccafcfcbfa4163d0c37437c21b&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/adsabs/resolver_service/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/adsabs/resolver_service/commit/&amp;amp;lt;a class=&amp;amp;quot;double-link&amp;amp;quot; href=&amp;amp;quot;https://git&amp;quot;&amp;gt;&amp;amp;lt;a class=&amp;lt;/a&amp;gt;hub.com/adsabs/resolver_service/commit/d636836228d121ccafcfcbfa4163d0c37437c21b&amp;quot;&amp;gt;d63683622&amp;lt;/a&amp;gt;&amp;lt;a href=&amp;quot;https://github.com/adsabs/resolver_service/commit/d636836228d121ccafcfcbfa4163d0c37437c21b&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/adsabs/resolver_service/commit/d636836228d121ccafcfcbfa4163d0c3&amp;lt;/a&amp;gt;7437c21b&amp;qu</a>ot;&gt;&amp;quot;&amp;gt;Merge &amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/adsabs/resolver_service/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/adsabs/resolver_service/commit/&lt;a class=&quot;double-link&quot; href=&quot;https:/... (continued)

Pull Request Pull Request #86: addition for db 2.0

Run Details

373 of 373 new or added lines in 3 files covered. (100.0%)

740 of 804 relevant lines covered (92.04%)

0.92 hits per line

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

93.66

/resolversrv/utils.py


from builtins import str
from builtins import range
from flask import current_app
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy import and_
from sqlalchemy.dialects.postgresql import insert

from google.protobuf import json_format

from resolversrv.models import DataLinks, Documents

def get_records(bibcode, link_type=None, link_sub_type=None):
    """
    Queries nonbib.datalinks table and returns results.

    :param bibcode:
    :param link_type:
    :param link_sub_type:
    :return: list of json records or None
    """
    with current_app.session_scope() as session:
        if link_type is None:
            rows = session.query(DataLinks).filter(and_(DataLinks.bibcode == bibcode)).all()
            current_app.logger.info("Fetched records for bibcode = %s."  % (bibcode))
        elif link_sub_type is None:
            rows = session.query(DataLinks).filter(and_(DataLinks.bibcode == bibcode, DataLinks.link_type == link_type)).all()
            current_app.logger.info("Fetched records for bibcode = %s and link type = %s." % (bibcode, link_type))
        elif '%' in link_sub_type:
            rows = session.query(DataLinks).filter(and_(DataLinks.bibcode == bibcode, DataLinks.link_type == link_type,
                                                        DataLinks.link_sub_type.match(link_sub_type))).all()
            current_app.logger.info("Fetched records for bibcode = %s, link type = %s and link sub type = %s." %
                                    (bibcode, link_type, link_sub_type))
        else:
            rows = session.query(DataLinks).filter(and_(DataLinks.bibcode == bibcode, DataLinks.link_type == link_type,
                                                        DataLinks.link_sub_type == link_sub_type)).all()
            current_app.logger.info("Fetched records for bibcode = %s, link type = %s and link sub type = %s." %
                                    (bibcode, link_type, link_sub_type))

        if len(rows) == 0:
            if link_type is None:
                current_app.logger.error("No records found for bibcode = %s." % (bibcode))
            elif link_sub_type is None:
                current_app.logger.error("No records found for bibcode = %s and link type = %s." % (bibcode, link_type))
            else:
                current_app.logger.error("No records found for bibcode = %s, link type = %s and link sub type = %s." %
                                         (bibcode, link_type, link_sub_type))
            return None

        results = []
        for row in rows:
            results.append(row.toJSON())
        return results
    return None


def get_records_new(bibcode, link_type=None, link_sub_type=None):
    """
    Queries nonbib.documents table and returns results.

    :param bibcode:
    :param link_type:
    :param link_sub_type:
    :return: list of json records or None
    """
    with current_app.session_scope() as session:
        # just bibcode was passed in
        if link_type is None:
            row = session.query(Documents).filter(Documents.bibcode == bibcode).first()
            current_app.logger.info("Fetched records for bibcode = %s."  % (bibcode))
        # query on bibcocd and link_type
        elif link_sub_type is None:
            row = session.query(Documents.links[link_type].label('links')).filter(Documents.bibcode == bibcode).first()
            if row:
                row = Documents(bibcode, [], {link_type: row[0]})
            current_app.logger.info("Fetched records for bibcode = %s and link type = %s." % (bibcode, link_type))
        # query on all three bibcode, link_type, and link_sub_type (with or without wildcard)
        elif '%' in link_sub_type:
            # get all the records for link_type and go through them to determine if they match wildcard string
            row = session.query(Documents.links[link_type].label('links')).filter(Documents.bibcode == bibcode).first()
            partial_link_sub_type = link_sub_type.replace('%', '')
            try:
                link_sub_type_wildcard = {}
                for key, value in row[0].items():
                    if partial_link_sub_type in key:
                        link_sub_type_wildcard.update({key: value})
                if len(link_sub_type_wildcard) > 0:
                    row = Documents(bibcode, [], {link_type: link_sub_type_wildcard})
                else:
                    row = {}
            except (KeyError, AttributeError):
                row = {}
            current_app.logger.info("Fetched records for bibcode = %s, link type = %s and link sub type = %s." %
                                    (bibcode, link_type, link_sub_type))
        else:
            row = session.query(Documents.links[link_type][link_sub_type].label('links')).filter(Documents.bibcode == bibcode).first()
            if row:
                row = Documents(bibcode, [], {link_type: {link_sub_type: row[0]}})
            current_app.logger.info("Fetched records for bibcode = %s, link type = %s and link sub type = %s." %
                                    (bibcode, link_type, link_sub_type))

        if not row:
            if link_type is None:
                current_app.logger.error("No records found for bibcode = %s." % (bibcode))
            elif link_sub_type is None:
                current_app.logger.error("No records found for bibcode = %s and link type = %s." % (bibcode, link_type))
            else:
                current_app.logger.error("No records found for bibcode = %s, link type = %s and link sub type = %s." %
                                         (bibcode, link_type, link_sub_type))
            return None

        return row.toJSON()
    return None


def add_records(datalinks_records_list):
    """
    upserts records into db

    :param datalinks_records_list:
    :return: success boolean, plus a status text for retuning error message, if any, to the calling program
    """
    rows = []
    for i in range(len(datalinks_records_list.datalinks_records)):
        for j in range(len(datalinks_records_list.datalinks_records[i].data_links_rows)):
            rows.append([
                datalinks_records_list.datalinks_records[i].bibcode,
                datalinks_records_list.datalinks_records[i].data_links_rows[j].link_type,
                datalinks_records_list.datalinks_records[i].data_links_rows[j].link_sub_type,
                datalinks_records_list.datalinks_records[i].data_links_rows[j].url,
                datalinks_records_list.datalinks_records[i].data_links_rows[j].title,
                datalinks_records_list.datalinks_records[i].data_links_rows[j].item_count
            ])

    if len(rows) > 0:
        table = DataLinks.__table__
        stmt = insert(table).values(rows)

        no_update_cols = []
        update_cols = [c.name for c in table.c
                       if c not in list(table.primary_key.columns)
                       and c.name not in no_update_cols]

        on_conflict_stmt = stmt.on_conflict_do_update(
            index_elements=table.primary_key.columns,
            set_={k: getattr(stmt.excluded, k) for k in update_cols}
        )

        try:
            with current_app.session_scope() as session:
                session.execute(on_conflict_stmt)
            current_app.logger.info('updated db with new data successfully')
            return True, 'updated db with new data successfully'
        except SQLAlchemyError as e:
            current_app.logger.error('SQLAlchemy: ' + str(e))
            return False, 'SQLAlchemy: ' + str(e)
    return False, 'unable to add records to the database'


def add_records_new(documents):
    """
    upserts records into db

    :param documents:
    :return: success boolean, plus a status text for retuning error message, if any, to the calling program
    """
    rows = []
    for doc in documents.document_records:
        rows.append({"bibcode":doc.bibcode,
                     "identifier": doc.identifier,
                     "links": json_format.MessageToDict(doc.links)})

    if len(rows) > 0:
        table = Documents.__table__
        stmt = insert(table).values(rows)

        # get list of fields making up primary key
        primary_keys = [c.name for c in list(table.primary_key.columns)]
        # define dict of non-primary keys for updating
        update_dict = {c.name: c for c in stmt.excluded if not c.primary_key}

        on_conflict_stmt = stmt.on_conflict_do_update(index_elements=primary_keys, set_=update_dict)

        try:
            with current_app.session_scope() as session:
                session.execute(on_conflict_stmt)
            return True, 'updated db with new data successfully'
        except SQLAlchemyError as e:
            session.rollback()
            return False, 'SQLAlchemy: ' + str(e)
    return False, 'unable to add records to the database'


def del_records(bibcode_list):
    """

    :param bibcode_list:
    :return:
    """
    try:
        with current_app.session_scope() as session:
            count = 0
            for bibcode in bibcode_list:
                count += session.query(DataLinks).filter(DataLinks.bibcode == bibcode).delete(synchronize_session=False)
            session.commit()
    except SQLAlchemyError as e:
        session.rollback()
        current_app.logger.error('SQLAlchemy: ' + str(e))
        return False, 'SQLAlchemy: ' + str(e)
    return True, count, 'removed ' + str(count) + ' records of ' + str(len(bibcode_list)) + ' bibcodes'


def del_records_new(bibcode_list):
    """

    :param bibcode_list:
    :return:
    """
    try:
        with current_app.session_scope() as session:
            count = 0
            for bibcode in bibcode_list:
                count += session.query(Documents).filter(Documents.bibcode == bibcode).delete(synchronize_session=False)
            session.commit()
    except SQLAlchemyError as e:
        session.rollback()
        current_app.logger.error('SQLAlchemy: ' + str(e))
        return False, 'SQLAlchemy: ' + str(e)
    return True, count, 'removed ' + str(count) + ' records of ' + str(len(bibcode_list)) + ' bibcodes'


def get_ids(identifier_list):
    """
    returns list of bibcodes and identifiers for the input identifier_list

    :param identifier_list:
    :return:
    """
    query = "SELECT bibcode,identifier FROM documents WHERE identifier && '{\"%s\"}';"%('","'.join(identifier_list))
    with current_app.session_scope() as session:
        rows = session.execute(query)
        if rows.rowcount > 0:
            message = "Fetched %d records for %d inputted identifiers." % (len(identifier_list), rows.rowcount)
            current_app.logger.info(message)
            return [dict(row) for row in rows], message
        current_app.logger.error('no matches found in database')
    return None, 'no matches found in database'

1
2	from builtins import str	1✔
3	from builtins import range	1✔
4	from flask import current_app	1✔
5	from sqlalchemy.exc import SQLAlchemyError	1✔
6	from sqlalchemy import and_	1✔
7	from sqlalchemy.dialects.postgresql import insert	1✔
8
9	from google.protobuf import json_format	1✔
10
11	from resolversrv.models import DataLinks, Documents	1✔
12
13	def get_records(bibcode, link_type=None, link_sub_type=None):	1✔
14	"""
15	Queries nonbib.datalinks table and returns results.
16
17	:param bibcode:
18	:param link_type:
19	:param link_sub_type:
20	:return: list of json records or None
21	"""
22	with current_app.session_scope() as session:	1✔
23	if link_type is None:	1✔
24	rows = session.query(DataLinks).filter(and_(DataLinks.bibcode == bibcode)).all()	1✔
25	current_app.logger.info("Fetched records for bibcode = %s." % (bibcode))	1✔
26	elif link_sub_type is None:	1✔
27	rows = session.query(DataLinks).filter(and_(DataLinks.bibcode == bibcode, DataLinks.link_type == link_type)).all()	1✔
28	current_app.logger.info("Fetched records for bibcode = %s and link type = %s." % (bibcode, link_type))	1✔
29	elif '%' in link_sub_type:	1✔
30	rows = session.query(DataLinks).filter(and_(DataLinks.bibcode == bibcode, DataLinks.link_type == link_type,	1✔
31	DataLinks.link_sub_type.match(link_sub_type))).all()
32	current_app.logger.info("Fetched records for bibcode = %s, link type = %s and link sub type = %s." %	1✔
33	(bibcode, link_type, link_sub_type))
34	else:
35	rows = session.query(DataLinks).filter(and_(DataLinks.bibcode == bibcode, DataLinks.link_type == link_type,	1✔
36	DataLinks.link_sub_type == link_sub_type)).all()
37	current_app.logger.info("Fetched records for bibcode = %s, link type = %s and link sub type = %s." %	1✔
38	(bibcode, link_type, link_sub_type))
39
40	if len(rows) == 0:	1✔
41	if link_type is None:	1✔
42	current_app.logger.error("No records found for bibcode = %s." % (bibcode))	1✔
43	elif link_sub_type is None:	1✔
44	current_app.logger.error("No records found for bibcode = %s and link type = %s." % (bibcode, link_type))	1✔
45	else:
46	current_app.logger.error("No records found for bibcode = %s, link type = %s and link sub type = %s." %	1✔
47	(bibcode, link_type, link_sub_type))
48	return None	1✔
49
50	results = []	1✔
51	for row in rows:	1✔
52	results.append(row.toJSON())	1✔
53	return results	1✔
54	return None	×
55
56
57	def get_records_new(bibcode, link_type=None, link_sub_type=None):	1✔
58	"""
59	Queries nonbib.documents table and returns results.
60
61	:param bibcode:
62	:param link_type:
63	:param link_sub_type:
64	:return: list of json records or None
65	"""
66	with current_app.session_scope() as session:	1✔
67	# just bibcode was passed in
68	if link_type is None:	1✔
69	row = session.query(Documents).filter(Documents.bibcode == bibcode).first()	1✔
70	current_app.logger.info("Fetched records for bibcode = %s." % (bibcode))	1✔
71	# query on bibcocd and link_type
72	elif link_sub_type is None:	1✔
73	row = session.query(Documents.links[link_type].label('links')).filter(Documents.bibcode == bibcode).first()	1✔
74	if row:	1✔
75	row = Documents(bibcode, [], {link_type: row[0]})	1✔
76	current_app.logger.info("Fetched records for bibcode = %s and link type = %s." % (bibcode, link_type))	1✔
77	# query on all three bibcode, link_type, and link_sub_type (with or without wildcard)
78	elif '%' in link_sub_type:	1✔
79	# get all the records for link_type and go through them to determine if they match wildcard string
80	row = session.query(Documents.links[link_type].label('links')).filter(Documents.bibcode == bibcode).first()	1✔
81	partial_link_sub_type = link_sub_type.replace('%', '')	1✔
82	try:	1✔
83	link_sub_type_wildcard = {}	1✔
84	for key, value in row[0].items():	1✔
85	if partial_link_sub_type in key:	1✔
86	link_sub_type_wildcard.update({key: value})	1✔
87	if len(link_sub_type_wildcard) > 0:	1✔
88	row = Documents(bibcode, [], {link_type: link_sub_type_wildcard})	1✔
89	else:
90	row = {}	1✔
91	except (KeyError, AttributeError):	1✔
92	row = {}	1✔
93	current_app.logger.info("Fetched records for bibcode = %s, link type = %s and link sub type = %s." %	1✔
94	(bibcode, link_type, link_sub_type))
95	else:
96	row = session.query(Documents.links[link_type][link_sub_type].label('links')).filter(Documents.bibcode == bibcode).first()	1✔
97	if row:	1✔
98	row = Documents(bibcode, [], {link_type: {link_sub_type: row[0]}})	1✔
99	current_app.logger.info("Fetched records for bibcode = %s, link type = %s and link sub type = %s." %	1✔
100	(bibcode, link_type, link_sub_type))
101
102	if not row:	1✔
103	if link_type is None:	1✔
104	current_app.logger.error("No records found for bibcode = %s." % (bibcode))	1✔
105	elif link_sub_type is None:	1✔
106	current_app.logger.error("No records found for bibcode = %s and link type = %s." % (bibcode, link_type))	1✔
107	else:
108	current_app.logger.error("No records found for bibcode = %s, link type = %s and link sub type = %s." %	1✔
109	(bibcode, link_type, link_sub_type))
110	return None	1✔
111
112	return row.toJSON()	1✔
113	return None	×
114
115
116	def add_records(datalinks_records_list):	1✔
117	"""
118	upserts records into db
119
120	:param datalinks_records_list:
121	:return: success boolean, plus a status text for retuning error message, if any, to the calling program
122	"""
123	rows = []	1✔
124	for i in range(len(datalinks_records_list.datalinks_records)):	1✔
125	for j in range(len(datalinks_records_list.datalinks_records[i].data_links_rows)):	1✔
126	rows.append([	1✔
127	datalinks_records_list.datalinks_records[i].bibcode,
128	datalinks_records_list.datalinks_records[i].data_links_rows[j].link_type,
129	datalinks_records_list.datalinks_records[i].data_links_rows[j].link_sub_type,
130	datalinks_records_list.datalinks_records[i].data_links_rows[j].url,
131	datalinks_records_list.datalinks_records[i].data_links_rows[j].title,
132	datalinks_records_list.datalinks_records[i].data_links_rows[j].item_count
133	])
134
135	if len(rows) > 0:	1✔
136	table = DataLinks.__table__	1✔
137	stmt = insert(table).values(rows)	1✔
138
139	no_update_cols = []	1✔
140	update_cols = [c.name for c in table.c	1✔
141	if c not in list(table.primary_key.columns)
142	and c.name not in no_update_cols]
143
144	on_conflict_stmt = stmt.on_conflict_do_update(	1✔
145	index_elements=table.primary_key.columns,
146	set_={k: getattr(stmt.excluded, k) for k in update_cols}
147	)
148
149	try:	1✔
150	with current_app.session_scope() as session:	1✔
151	session.execute(on_conflict_stmt)	1✔
152	current_app.logger.info('updated db with new data successfully')	1✔
153	return True, 'updated db with new data successfully'	1✔
154	except SQLAlchemyError as e:	×
155	current_app.logger.error('SQLAlchemy: ' + str(e))	×
156	return False, 'SQLAlchemy: ' + str(e)	×
157	return False, 'unable to add records to the database'	1✔
158
159
160	def add_records_new(documents):	1✔
161	"""
162	upserts records into db
163
164	:param documents:
165	:return: success boolean, plus a status text for retuning error message, if any, to the calling program
166	"""
167	rows = []	1✔
168	for doc in documents.document_records:	1✔
169	rows.append({"bibcode":doc.bibcode,	1✔
170	"identifier": doc.identifier,
171	"links": json_format.MessageToDict(doc.links)})
172
173	if len(rows) > 0:	1✔
174	table = Documents.__table__	1✔
175	stmt = insert(table).values(rows)	1✔
176
177	# get list of fields making up primary key
178	primary_keys = [c.name for c in list(table.primary_key.columns)]	1✔
179	# define dict of non-primary keys for updating
180	update_dict = {c.name: c for c in stmt.excluded if not c.primary_key}	1✔
181
182	on_conflict_stmt = stmt.on_conflict_do_update(index_elements=primary_keys, set_=update_dict)	1✔
183
184	try:	1✔
185	with current_app.session_scope() as session:	1✔
186	session.execute(on_conflict_stmt)	1✔
187	return True, 'updated db with new data successfully'	1✔
188	except SQLAlchemyError as e:	1✔
189	session.rollback()	1✔
190	return False, 'SQLAlchemy: ' + str(e)	1✔
191	return False, 'unable to add records to the database'	1✔
192
193
194	def del_records(bibcode_list):	1✔
195	"""
196
197	:param bibcode_list:
198	:return:
199	"""
200	try:	1✔
201	with current_app.session_scope() as session:	1✔
202	count = 0	1✔
203	for bibcode in bibcode_list:	1✔
204	count += session.query(DataLinks).filter(DataLinks.bibcode == bibcode).delete(synchronize_session=False)	1✔
205	session.commit()	1✔
206	except SQLAlchemyError as e:	×
207	session.rollback()	×
208	current_app.logger.error('SQLAlchemy: ' + str(e))	×
209	return False, 'SQLAlchemy: ' + str(e)	×
210	return True, count, 'removed ' + str(count) + ' records of ' + str(len(bibcode_list)) + ' bibcodes'	1✔
211
212
213	def del_records_new(bibcode_list):	1✔
214	"""
215
216	:param bibcode_list:
217	:return:
218	"""
219	try:	1✔
220	with current_app.session_scope() as session:	1✔
221	count = 0	1✔
222	for bibcode in bibcode_list:	1✔
223	count += session.query(Documents).filter(Documents.bibcode == bibcode).delete(synchronize_session=False)	1✔
224	session.commit()	1✔
225	except SQLAlchemyError as e:	1✔
226	session.rollback()	1✔
227	current_app.logger.error('SQLAlchemy: ' + str(e))	1✔
228	return False, 'SQLAlchemy: ' + str(e)	1✔
229	return True, count, 'removed ' + str(count) + ' records of ' + str(len(bibcode_list)) + ' bibcodes'	1✔
230
231
232	def get_ids(identifier_list):	1✔
233	"""
234	returns list of bibcodes and identifiers for the input identifier_list
235
236	:param identifier_list:
237	:return:
238	"""
239	query = "SELECT bibcode,identifier FROM documents WHERE identifier && '{\"%s\"}';"%('","'.join(identifier_list))	1✔
240	with current_app.session_scope() as session:	1✔
241	rows = session.execute(query)	1✔
242	if rows.rowcount > 0:	1✔
243	message = "Fetched %d records for %d inputted identifiers." % (len(identifier_list), rows.rowcount)	1✔
244	current_app.logger.info(message)	1✔
245	return [dict(row) for row in rows], message	1✔
246	current_app.logger.error('no matches found in database')	1✔
247	return None, 'no matches found in database'	1✔

adsabs / resolver_service / 1481284286

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