20760795939

Committed 06 Jan 2026 08:14PM UTC coverage: 84.247%. First build

Build # 20760795939

Build Type

Pull #187

github

Committed by

web-flow

Commit Message

Merge 2a9120d73 into ca28750ca

Pull Request Pull Request #187: updating biblib to python 3.10

Run Details

76 of 88 new or added lines in 3 files covered. (86.36%)

1706 of 2025 relevant lines covered (84.25%)

0.84 hits per line

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

85.88

/biblib/cli.py

import click
from flask import current_app
from flask.cli import with_appcontext
from biblib.models import User, Permissions, Library, Notes
from datetime import datetime
from dateutil.relativedelta import relativedelta
import sqlalchemy_continuum

@click.group()
def biblib():
    """Commands for the biblib service"""
    pass

@biblib.command()
@with_appcontext
def syncdb():
    """
    Compares the users that exist within the API to those within the
    microservice and deletes any stale users that no longer exist. The logic
    also takes care of the associated permissions and libraries depending on
    the cascade that has been implemented.
    """
    with current_app.session_scope() as session:
        # Obtain the list of API users
        postgres_search_text = 'SELECT id FROM users;'
        result = session.execute(postgres_search_text).fetchall()
        list_of_api_users = [int(r[0]) for r in result]

        # Loop through every use in the service database
        removal_list = []
        for service_user in session.query(User).all():
            if service_user.absolute_uid not in list_of_api_users:
                try:
                    # Obtain the libraries that should be deleted
                    permissions = session.query(Permissions).filter(
                        Permissions.user_id == service_user.id
                    ).all()
                    
                    libraries = [
                        session.query(Library).filter(Library.id == p.library_id).one() 
                        for p in permissions if p.permissions['owner']
                    ]

                    # Delete all the libraries found
                    # By cascade this should delete all the permissions
                    for library in libraries:
                        session.delete(library)
                    for permission in permissions:
                        session.delete(permission)
                    
                    session.delete(service_user)
                    session.commit()
                    
                    d_len = len(libraries)
                    current_app.logger.info('Removed stale user: {} and {} libraries'.format(service_user, d_len))
                    
                    removal_list.append(service_user)
                    
                except Exception as error:
                    current_app.logger.info('Problem with database, could not remove user {}: {}'
                                            .format(service_user, error))
                    session.rollback()
        
        current_app.logger.info('Deleted {} stale users: {}'.format(len(removal_list), removal_list))

@biblib.command(name='clean_versions_time')
@click.option('--years', default=None, type=int, help='Number of years to keep')
@with_appcontext
def clean_versions_time(years):
    """
    Clears obsolete library and notes versions older than chosen time.
    """
    if not years:
        years = current_app.config.get('REVISION_TIME', 7)

    with current_app.session_scope() as session:
        # Obtain a list of all versions older than chosen time.
        LibraryVersion = sqlalchemy_continuum.version_class(Library)
        NotesVersion = sqlalchemy_continuum.version_class(Notes)

        current_date = datetime.now()
        current_offset = current_date - relativedelta(years=years)

        try:
            library_results = session.query(LibraryVersion).filter(
                LibraryVersion.date_last_modified < current_offset
            ).all()
            notes_results = session.query(NotesVersion).filter(
                NotesVersion.date_last_modified < current_offset
            ).all()

            for revision in library_results:
                session.delete(revision)
            for revision in notes_results:
                session.delete(revision)

            session.commit()

            current_app.logger.info('Removed {} obsolete library revisions'.format(len(library_results)))
            current_app.logger.info('Removed {} obsolete notes revisions'.format(len(notes_results)))

        except Exception as error:
            current_app.logger.info('Problem with database, could not remove revisions: {}'
                                    .format(error))
            session.rollback()

@biblib.command(name='clean_versions_number')
@click.option('--revisions', default=None, type=int, help='Maximum revisions to keep')
@with_appcontext
def clean_versions_number(revisions):
    """
    Limits number of revisions saved per entity to n_revisions.
    """
    if not revisions:
        revisions = current_app.config.get('NUMBER_REVISIONS', 7)

    def limit_revisions(session, entity_class, n_revisions):
        VersionClass = sqlalchemy_continuum.version_class(entity_class)
        entities = session.query(entity_class).all()

        for entity in entities:
            try:
                revs = session.query(VersionClass).filter_by(id=entity.id).order_by(
                    VersionClass.date_last_modified.asc()
                ).all()
                
                current_app.logger.debug('Found {} revisions for entity: {}'.format(len(revs), entity.id))
                
                if len(revs) > n_revisions:
                    obsolete = revs[:-n_revisions]
                    for r in obsolete:
                        session.delete(r)
                    
                    session.commit()
                    
                    current_app.logger.info('Removed {} obsolete revisions for entity: {}'.format(len(obsolete), entity.id))
                    
            except Exception as error:
                current_app.logger.info('Problem with the database, could not remove revisions for entity {}: {}'
                                        .format(entity, error))
                session.rollback()

    with current_app.session_scope() as session:
        limit_revisions(session, Library, revisions)
        limit_revisions(session, Notes, revisions)

1	import click	1✔
2	from flask import current_app	1✔
3	from flask.cli import with_appcontext	1✔
4	from biblib.models import User, Permissions, Library, Notes	1✔
5	from datetime import datetime	1✔
6	from dateutil.relativedelta import relativedelta	1✔
7	import sqlalchemy_continuum	1✔
8
9	@click.group()	1✔
10	def biblib():	1✔
11	"""Commands for the biblib service"""
NEW 12	pass	×
13
14	@biblib.command()	1✔
15	@with_appcontext	1✔
16	def syncdb():	1✔
17	"""
18	Compares the users that exist within the API to those within the
19	microservice and deletes any stale users that no longer exist. The logic
20	also takes care of the associated permissions and libraries depending on
21	the cascade that has been implemented.
22	"""
23	with current_app.session_scope() as session:	1✔
24	# Obtain the list of API users
25	postgres_search_text = 'SELECT id FROM users;'	1✔
26	result = session.execute(postgres_search_text).fetchall()	1✔
27	list_of_api_users = [int(r[0]) for r in result]	1✔
28
29	# Loop through every use in the service database
30	removal_list = []	1✔
31	for service_user in session.query(User).all():	1✔
32	if service_user.absolute_uid not in list_of_api_users:	1✔
33	try:	1✔
34	# Obtain the libraries that should be deleted
35	permissions = session.query(Permissions).filter(	1✔
36	Permissions.user_id == service_user.id
37	).all()
38
39	libraries = [	1✔
40	session.query(Library).filter(Library.id == p.library_id).one()
41	for p in permissions if p.permissions['owner']
42	]
43
44	# Delete all the libraries found
45	# By cascade this should delete all the permissions
46	for library in libraries:	1✔
47	session.delete(library)	1✔
48	for permission in permissions:	1✔
49	session.delete(permission)	1✔
50
51	session.delete(service_user)	1✔
52	session.commit()	1✔
53
54	d_len = len(libraries)	1✔
55	current_app.logger.info('Removed stale user: {} and {} libraries'.format(service_user, d_len))	1✔
56
57	removal_list.append(service_user)	1✔
58
NEW 59	except Exception as error:	×
NEW 60	current_app.logger.info('Problem with database, could not remove user {}: {}'	×
61	.format(service_user, error))
NEW 62	session.rollback()	×
63
64	current_app.logger.info('Deleted {} stale users: {}'.format(len(removal_list), removal_list))	1✔
65
66	@biblib.command(name='clean_versions_time')	1✔
67	@click.option('--years', default=None, type=int, help='Number of years to keep')	1✔
68	@with_appcontext	1✔
69	def clean_versions_time(years):	1✔
70	"""
71	Clears obsolete library and notes versions older than chosen time.
72	"""
73	if not years:	1✔
NEW 74	years = current_app.config.get('REVISION_TIME', 7)	×
75
76	with current_app.session_scope() as session:	1✔
77	# Obtain a list of all versions older than chosen time.
78	LibraryVersion = sqlalchemy_continuum.version_class(Library)	1✔
79	NotesVersion = sqlalchemy_continuum.version_class(Notes)	1✔
80
81	current_date = datetime.now()	1✔
82	current_offset = current_date - relativedelta(years=years)	1✔
83
84	try:	1✔
85	library_results = session.query(LibraryVersion).filter(	1✔
86	LibraryVersion.date_last_modified < current_offset
87	).all()
88	notes_results = session.query(NotesVersion).filter(	1✔
89	NotesVersion.date_last_modified < current_offset
90	).all()
91
92	for revision in library_results:	1✔
93	session.delete(revision)	1✔
94	for revision in notes_results:	1✔
95	session.delete(revision)	1✔
96
97	session.commit()	1✔
98
99	current_app.logger.info('Removed {} obsolete library revisions'.format(len(library_results)))	1✔
100	current_app.logger.info('Removed {} obsolete notes revisions'.format(len(notes_results)))	1✔
101
NEW 102	except Exception as error:	×
NEW 103	current_app.logger.info('Problem with database, could not remove revisions: {}'	×
104	.format(error))
NEW 105	session.rollback()	×
106
107	@biblib.command(name='clean_versions_number')	1✔
108	@click.option('--revisions', default=None, type=int, help='Maximum revisions to keep')	1✔
109	@with_appcontext	1✔
110	def clean_versions_number(revisions):	1✔
111	"""
112	Limits number of revisions saved per entity to n_revisions.
113	"""
114	if not revisions:	1✔
NEW 115	revisions = current_app.config.get('NUMBER_REVISIONS', 7)	×
116
117	def limit_revisions(session, entity_class, n_revisions):	1✔
118	VersionClass = sqlalchemy_continuum.version_class(entity_class)	1✔
119	entities = session.query(entity_class).all()	1✔
120
121	for entity in entities:	1✔
122	try:	1✔
123	revs = session.query(VersionClass).filter_by(id=entity.id).order_by(	1✔
124	VersionClass.date_last_modified.asc()
125	).all()
126
127	current_app.logger.debug('Found {} revisions for entity: {}'.format(len(revs), entity.id))	1✔
128
129	if len(revs) > n_revisions:	1✔
130	obsolete = revs[:-n_revisions]	1✔
131	for r in obsolete:	1✔
132	session.delete(r)	1✔
133
134	session.commit()	1✔
135
136	current_app.logger.info('Removed {} obsolete revisions for entity: {}'.format(len(obsolete), entity.id))	1✔
137
NEW 138	except Exception as error:	×
NEW 139	current_app.logger.info('Problem with the database, could not remove revisions for entity {}: {}'	×
140	.format(entity, error))
NEW 141	session.rollback()	×
142
143	with current_app.session_scope() as session:	1✔
144	limit_revisions(session, Library, revisions)	1✔
145	limit_revisions(session, Notes, revisions)	1✔

adsabs / biblib-service / 20760795939

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