Tag refactoring

[udgover]

Following a recent discussion about tags, I'm creating one here so that we do not lose any context.

The idea is to no longer rely on tags graph to handle tags in Yeti. There will still be a tags collection to store all existing tags across Yeti but then another model is used to handle tags for all yeti objects.

The tags collection would look like this:

class Tag(YetiArangoConnector):
   _collection = "tags"
   name: str
   expiration: datetime.timedelta = datetime.timedelta(yeti_config.get("default_tag_expiration")

Then we would have the following model to handle tagged objects

class Tagged(BaseModel):
   name: str
   tagged: datetime.datetime
   expired: bool = False

class TaggedList(BaseModel):
   tags = List[Tagged]

class YetiTagModel(BaseModel):
   _tags = TaggedList

   def __init__(self, **data):
      if "tags" in data:
         self._tags = TaggedList(data["tags])
         del data["tags"]
      super().__init__(**data)

   @computed_field
   def tags(self) -> TaggedList:
      return self._tags

   def has_tag(self, tag: str):
      for t in self._tags:
         if t.name == tag:
            return True
       return False

   def get_tag(self, tag: str):
       for t in self._tags:
          if t.name == tag:
             return t
        return None

   def add_tag(self, tag: str) -> None:
      if new_tag := self.get_tag(tag):
         new_tag.tagged = datetime.now()
      else:
         new_tag = Tagged(tag)
         self._tags.append(new_tag)
      // add / update tags collection
      if not Tag.find(tag):
         Tag(tag).save()

   def add_tags(self, tags: List[str]):
      for t in tags:
        self.add_tag(t)
   
   def refresh_tag(self, tag: str):
       if tag := self.get_tag(tag):
          tag.tagged = datetime.now()
       else:
          raise ValueError(f"Tag {tag} does not exists")

We could even handle the source of the tags:

class Tagged(BaseModel):
   name: str
   tagged: datetime.datetime
   expired: bool = False
   sources: List[str]
 
   def add_tag(tag: str, source: str = "manual"):
      ...
 
   def add_tags(tags: List[str], source: str = "manual"):
      ...

In order to smoothen the migration, we could also provide a migration script like:

def migrate():
   db = ArangoClient()
   if not db.has_graph('tags'):
      print("Already migrated")
   for tag in db.graph('tags'):
      for target in tags.edges:
         target.add_tag(tag.name)
         Tag(tag.name).save()
      tag.edges.delete()
   db.graph.delete('tags')