adding the notes table and versioning

alembic/env.py

@@ -2,6 +2,7 @@
 from alembic import context
 from sqlalchemy import engine_from_config, pool
 from logging.config import fileConfig
+import biblib.models
 import os
 import sys
 
@@ -20,7 +21,7 @@
 #from flask import current_app
 #config.set_main_option('sqlalchemy.url', current_app.config.get('SQLALCHEMY_BINDS')['libraries'])
 #target_metadata = current_app.extensions['migrate'].db.metadata
-target_metadata = None
+target_metadata = biblib.models.Base.metadata
 
 # other values from the config, defined by the needs of env.py,
 # can be acquired:

alembic/versions/8feb0a6fbd27_adding_the_notes_table_and_changing_.py

@@ -0,0 +1,35 @@
+"""Adding the notes table and changing libraries table
+
+Revision ID: 8feb0a6fbd27
+Revises: 4039aa74290f
+Create Date: 2023-08-13 21:38:18.660449
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = '8feb0a6fbd27'
+down_revision = '4039aa74290f'
+
+from alembic import op
+import sqlalchemy as sa
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('notes',
+    sa.Column('id', sa.Integer(), nullable=False),
+    sa.Column('content', sa.UnicodeText(), nullable=True),
+    sa.Column('bibcode', sa.String(length=50), nullable=False),
+    sa.Column('library_id', biblib.models.GUID(), nullable=True),
+    sa.Column('date_created', sa.DateTime(), nullable=False),
+    sa.Column('date_last_modified', sa.DateTime(), nullable=False),
+    sa.ForeignKeyConstraint(['library_id'], ['library.id'], ),
+    sa.PrimaryKeyConstraint('id')
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table('notes')
+    # ### end Alembic commands ###

biblib/manage.py

@@ -11,7 +11,7 @@
 from flask import current_app
 from flask_script import Manager, Command, Option
 from flask_migrate import Migrate, MigrateCommand
-from biblib.models import Base, User, Permissions, Library
+from biblib.models import Base, User, Permissions, Library, Notes
 from biblib.app import create_app
 from sqlalchemy import create_engine, desc
 from sqlalchemy.orm import sessionmaker, scoped_session
@@ -67,7 +67,7 @@ def run(app=app):
 
 class DeleteObsoleteVersionsTime(Command):
     """
-    Clears obsolete library versions older than chosen time.
+    Clears obsolete library and notes versions older than chosen time.
     """
     @staticmethod
     def run(app=app, n_years=None):
@@ -81,50 +81,61 @@ def run(app=app, n_years=None):
             with current_app.session_scope() as session:
                 # Obtain a list of all versions older than 1 year.
                 LibraryVersion = sqlalchemy_continuum.version_class(Library)
+                NotesVersion = sqlalchemy_continuum.version_class(Notes)
                 current_date = datetime.now()
                 current_offset = current_date - relativedelta(years=n_years)
                 try:
-                    results = session.query(LibraryVersion).filter(LibraryVersion.date_last_modified<current_offset).all()
-                    d = [session.delete(revision) for revision in results]
-                    d_len = len(d)
+                    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() 
+
+                    d_library = [session.delete(revision) for revision in library_results]
+                    d_notes = [session.delete(revision) for revision in notes_results] 
+
+                    d_library_len = len(d_library)
+                    d_notes_len = len(d_notes)
                     session.commit()
-                    current_app.logger.info('Removed {} obsolete revisions'.format(d_len))
+                    current_app.logger.info('Removed {} obsolete library revisions'.format(d_library_len))
+                    current_app.logger.info('Removed {} obsolete notes revisions'.format(d_notes_len))
                 except Exception as error:
                         current_app.logger.info('Problem with database, could not remove revisions: {}'
                                                 .format(error))
                         session.rollback()
 
 class DeleteObsoleteVersionsNumber(Command):
     """
-    Limits number of revisions saved per library to n_revisions.
+    Limits number of revisions saved per entity to n_revisions.
     """
+    @staticmethod
+    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:
+                revisions = 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(revisions), entity.id))
+                obsolete_revisions = revisions[:-n_revisions]
+                deleted_revisions = [session.delete(revision) for revision in obsolete_revisions]
+                deleted_revisions_len = len(deleted_revisions)
+                session.commit()
+                current_app.logger.info('Removed {} obsolete revisions for entity: {}'.format(deleted_revisions_len, 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()
+
     @staticmethod
     def run(app=app, n_revisions=None):
-        """
-        Carries out the deletion of older versions
-        """
-        if not n_revisions: n_revisions = current_app.config.get('NUMBER_REVISIONS', 7)
+        if not n_revisions:
+            n_revisions = current_app.config.get('NUMBER_REVISIONS', 7)
 
         with app.app_context():
             with current_app.session_scope() as session:
-                LibraryVersion = sqlalchemy_continuum.version_class(Library)
-                for library in session.query(Library).all():
-                    try:
-                        #for library in libraries:
-                        revisions = session.query(LibraryVersion).filter_by(id=library.id).order_by(LibraryVersion.date_last_modified.asc()).all()
-                        # Obtain the revisions for a given library
-                        current_app.logger.debug('Found {} revisions for library: {}'.format(len(revisions), library.id))
-                        obsolete_revisions = revisions[:-n_revisions]
-                        d = [session.delete(revision) for revision in obsolete_revisions]
-                        #deletes all but the n_revisions most recent revisions.
-                        d_len = len(d)
-                        session.commit()
-                        current_app.logger.info('Removed {} obsolete revisions for library: {}'.format(d_len, library.id))
-
-                    except Exception as error:
-                        current_app.logger.info('Problem with database, could not remove revisions for library {}: {}'
-                                                .format(library, error))
-                        session.rollback()
+                DeleteObsoleteVersionsNumber.limit_revisions(session, Library, n_revisions)
+                DeleteObsoleteVersionsNumber.limit_revisions(session, Notes, n_revisions)
+
+
 
 # Setup the command line arguments using Flask-Script
 manager = Manager(app)

biblib/models.py

@@ -11,7 +11,7 @@
 from sqlalchemy.ext.mutable import Mutable
 from sqlalchemy.types import TypeDecorator, CHAR, String as StringType
 from sqlalchemy.ext.declarative import declarative_base
-from sqlalchemy import Column, Integer, String, ForeignKey, DateTime, Boolean
+from sqlalchemy import Column, Integer, String, ForeignKey, DateTime, Boolean, UnicodeText, UniqueConstraint
 from sqlalchemy.orm import relationship, configure_mappers
 from sqlalchemy_continuum import make_versioned
 make_versioned(user_cls=None)
@@ -172,7 +172,52 @@ def __repr__(self):
         return '<User {0}, {1}>'\
             .format(self.id, self.absolute_uid)
 
+class Notes(Base): 
+    """ 
+    Notes table 
+    """
+
+    __tablename__ = 'notes'
+    versioned = {}
+    id = Column(Integer, primary_key=True)
+    content = Column(UnicodeText)
+    bibcode = Column(String(50), nullable=False)
+    library_id = Column(GUID, ForeignKey('library.id'))
+    date_created = Column(
+        DateTime,
+        nullable=False,
+        default=datetime.utcnow
+    )
+    date_last_modified = Column(
+        DateTime,
+        nullable=False,
+        default=datetime.utcnow,
+        onupdate=datetime.utcnow
+    )
 
+    def __repr__(self):
+        return '<Note, Note id: {0}, library: {1}, bibcode: {2}, ' \
+        'content: {3}>'.format(self.id, 
+                            self.library, 
+                            self.bibcode, 
+                            self.content)
+    @classmethod
+    def create_unique(cls, session, content, bibcode, library): 
+        """
+        Creates a new note in the database
+        """
+        if bibcode not in library.bibcode.keys(): 
+            raise ValueError('Bibcode {0} not in library {1}'.format(bibcode, library))
+        existing_note = session.query(Notes).filter_by(bibcode=bibcode, library_id=library.id).first()
+
+        if existing_note:
+            raise ValueError("A note for the same bibcode and library already exists.")
+
+        note = Notes(content=content, bibcode=bibcode, library_id=library.id)
+        session.add(note)
+        session.commit()
+        return note
+
 class Library(Base):
     """
     Library table
@@ -186,6 +231,9 @@ class Library(Base):
     description = Column(String(200))
     public = Column(Boolean)
     bibcode = Column(MutableDict.as_mutable(JSON), default={})
+    notes = relationship('Notes',
+                            backref='library',
+                            cascade='all, delete-orphan')
     date_created = Column(
         DateTime,
         nullable=False,
@@ -200,7 +248,7 @@ class Library(Base):
     permissions = relationship('Permissions',
                                   backref='library',
                                   cascade='delete')
-
+    
     def __repr__(self):
         return '<Library, library_id: {0} name: {1}, ' \
                'description: {2}, public: {3},' \
@@ -235,15 +283,22 @@ def add_bibcodes(self, bibcodes):
 
     def remove_bibcodes(self, bibcodes):
         """
-        Removes a bibcode(s) from the bibcode field.
+        Removes a bibcode(s) and their associated notes from the bibcode field.
 
         Given the way in which bibcodes are stored may change, it seems simpler
         to keep the method of adding/removing in a small wrapper so that only
         one location needs to be modified (or YAGNI?).
 
         :param bibcodes: list of bibcodes
         """
-        [self.bibcode.pop(key, None) for key in bibcodes]
+
+        bibcode_to_notes = {note.bibcode: note for note in self.notes}
+
+        for bibcode in bibcodes: 
+            if bibcode in bibcode_to_notes: 
+                note = bibcode_to_notes[bibcode]
+                self.notes.remove(note)
+            self.bibcode.pop(bibcode, None)
 
 
 class Permissions(Base):

biblib/tests/unit_tests/test_models.py

@@ -3,10 +3,12 @@
 """
 
 import unittest
-from biblib.models import User, Library, Permissions, MutableDict
+from biblib.models import User, Library, Permissions, MutableDict, Notes
 from biblib.tests.base import TestCaseDatabase
+from sqlalchemy import exc
+import pytest
 
-class TestLibraryModel(TestCaseDatabase):
+class TestLibraryModel(TestCaseDatabase):                                                                                                                                                                                                                                                                     
     """
     Class for testing the methods usable by the Library model
     """
@@ -91,5 +93,111 @@ def test_coerce(self):
         same_list = mutable_dict.coerce('key', mutable_dict)
         self.assertEqual(same_list, mutable_dict)
 
+    def test_create_unique_note(self): 
+        lib = Library(bibcode={'1': {}, '2': {}, '3': {}}, public=True, description="Test description")
+        with self.app.session_scope() as session: 
+            session.add(lib) 
+            session.commit()
+
+            note1 = Notes.create_unique(session, content="Test content 1", bibcode="1", library=lib)
+            session.add(note1) 
+            session.commit() 
+
+            with self.assertRaises(ValueError) as context:
+                note2 = Notes.create_unique(session, content="Test content 2", bibcode="1", library=lib) 
+                session.add(note2)
+                session.commit()
+
+            existing_notes = session.query(Notes).filter_by(bibcode="1", library_id=lib.id).all() 
+            self.assertEqual(len(existing_notes), 1) 
+            self.assertEqual(existing_notes[0].content, "Test content 1") 
+            self.assertEqual(lib.notes, [note1])
+
+    def test_create_unique_bibcode_not_in_library(self):
+
+        lib = Library(bibcode={'1': {}, '2': {}}, public=True, description="Test description")
+        with self.app.session_scope() as session: 
+            session.add(lib) 
+            session.commit()
+
+            with self.assertRaises(ValueError) as context:
+                Notes.create_unique(session, content="Test content 1", bibcode="3", library=lib) 
+
+            self.assertUnsortedEqual(lib.get_bibcodes(), ['1', '2'])
+            self.assertEqual(lib.notes, [])
+            self.assertIn("Bibcode 3 not in library", context.exception.args[0])
+
+    def test_library_notes_relationship(self): 
+        lib = Library(bibcode={'1': {}, '2': {}}, public=True, description="Test description")
+
+        with self.app.session_scope() as session: 
+            session.add(lib) 
+            session.commit()
+            note1 = Notes.create_unique(session, content="Note 1 Content", bibcode="1", library=lib)
+            note2 = Notes.create_unique(session, content="Note 2 Content", bibcode="2", library=lib)
+
+            session.add_all([note1, note2]) 
+            session.commit()
+
+
+            self.assertEqual(lib.notes, [note1, note2])
+
+            session.delete(lib) 
+            session.commit() 
+
+            self.assertEqual(session.query(Notes).count(), 0)
+            self.assertEqual(session.query(Library).count(), 0)
+
+    def test_remove_bibcodes_remove_notes(self): 
+        lib = Library(bibcode={'1': {}, '2': {}}, public=True, description="Test description")
+
+        with self.app.session_scope() as session: 
+            session.add(lib) 
+            session.commit()
+            note1 = Notes.create_unique(session, content="Note 1 Content", bibcode="1", library=lib)
+            note2 = Notes.create_unique(session, content="Note 2 Content", bibcode="2", library=lib)
+
+            session.add(note1) 
+            session.add(note2)
+            session.commit()
+
+            self.assertEqual(lib.notes, [note1, note2])
+            self.assertEqual(session.query(Notes).count(), 2)
+
+
+            lib.remove_bibcodes(['1', '2'])
+
+            self.assertEqual(session.query(Notes).count(), 0)
+
+    def test_change_bibcodes_orphan_notes(self): 
+        lib = Library(bibcode={'1': {}, '2': {}}, public=True, description="Test description")
+
+        with self.app.session_scope() as session: 
+            session.add(lib) 
+            session.commit()
+            note1 = Notes.create_unique(session, content="Note 1 Content", bibcode="1", library=lib)
+
+            session.add(note1) 
+
+            session.commit()
+
+            self.assertEqual(lib.notes, [note1])
+            self.assertEqual(session.query(Notes).count(), 1)
+
+
+            lib.bibcode = {'2': {}, '3': {}}
+
+            self.assertEqual(lib.notes, [note1])
+            self.assertEqual(session.query(Notes).count(), 1)
+            self.assertEqual(lib.notes[0].bibcode, '1')
+
+
+
+
+
+
+
+
+
 if __name__ == '__main__':
     unittest.main(verbosity=2)