"""
Tags are entities that are attached to objects in the same way as
Attributes. But contrary to Attributes, which are unique to an
individual object, a single Tag can be attached to any number of
objects at the same time.
Tags are used for tagging, obviously, but the data structure is also
used for storing Aliases and Permissions. This module contains the
respective handlers.
"""
from collections import defaultdict
from django.conf import settings
from django.db import models
from evennia.locks.lockfuncs import perm as perm_lockfunc
from evennia.utils.utils import make_iter, to_str
_TYPECLASS_AGGRESSIVE_CACHE = settings.TYPECLASS_AGGRESSIVE_CACHE
# ------------------------------------------------------------
#
# Tags
#
# ------------------------------------------------------------
[docs]class Tag(models.Model):
"""
Tags are quick markers for objects in-game. An typeobject can have
any number of tags, stored via its db_tags property. Tagging
similar objects will make it easier to quickly locate the group
later (such as when implementing zones). The main advantage of
tagging as opposed to using tags is speed; a tag is very
limited in what data it can hold, and the tag key+category is
indexed for efficient lookup in the database. Tags are shared
between objects - a new tag is only created if the key+category
combination did not previously exist, making them unsuitable for
storing object-related data (for this a regular Attribute should be
used).
The 'db_data' field is intended as a documentation field for the
tag itself, such as to document what this tag+category stands for
and display that in a web interface or similar.
The main default use for Tags is to implement Aliases for objects.
this uses the 'aliases' tag category, which is also checked by the
default search functions of Evennia to allow quick searches by alias.
"""
db_key = models.CharField(
"key", max_length=255, null=True, help_text="tag identifier", db_index=True
)
db_category = models.CharField(
"category", max_length=64, null=True, blank=True, help_text="tag category", db_index=True
)
db_data = models.TextField(
"data",
null=True,
blank=True,
help_text="optional data field with extra information. This is not searched for.",
)
# this is "objectdb" etc. Required behind the scenes
db_model = models.CharField(
"model", max_length=32, null=True, help_text="database model to Tag", db_index=True
)
# this is None, alias or permission
db_tagtype = models.CharField(
"tagtype",
max_length=16,
null=True,
blank=True,
help_text="overall type of Tag",
db_index=True,
)
class Meta:
"Define Django meta options"
verbose_name = "Tag"
unique_together = (("db_key", "db_category", "db_tagtype", "db_model"),)
index_together = (("db_key", "db_category", "db_tagtype", "db_model"),)
def __lt__(self, other):
return str(self) < str(other)
def __str__(self):
return str(
"<Tag: %s%s>"
% (self.db_key, "(category:%s)" % self.db_category if self.db_category else "")
)
#
# Handlers making use of the Tags model
#
[docs]class TagProperty:
"""
Tag Property.
"""
taghandler_name = "tags"
[docs] def __init__(self, category=None, data=None):
"""
Tag property descriptor. Allows for setting tags on an object as Django-like 'fields'
on the class level. Since Tags are almost always used for querying, Tags are always
created/assigned along with the object. Make sure the property/tagname does not collide
with an existing method/property on the class. If it does, you must use tags.add()
instead.
Note that while you _can_ check e.g. `obj.tagname,this will give an AttributeError
if the Tag is not set. Most often you want to use `obj.tags.get("tagname")` to check
if a tag is set on an object.
Example:
::
class Character(DefaultCharacter):
mytag = TagProperty() # category=None
mytag2 = TagProperty(category="tagcategory")
"""
self._category = category
self._data = data
self._key = ""
def __set_name__(self, cls, name):
"""
Called when descriptor is first assigned to the class (not the instance!).
It is called with the name of the field.
"""
self._key = name
def __get__(self, instance, owner):
"""
Called when accessing the tag as a property on the instance.
"""
try:
return getattr(instance, self.taghandler_name).get(
key=self._key, category=self._category, return_list=False, raise_exception=True
)
except AttributeError:
self.__set__(instance, self._category)
def __set__(self, instance, category):
"""
Assign a new category to the tag. It's not possible to set 'data' this way.
"""
self._category = category
(
getattr(instance, self.taghandler_name).add(
key=self._key, category=self._category, data=self._data
)
)
def __delete__(self, instance):
"""
Called when running `del` on the property. Will disconnect the object from
the Tag. Note that the tag will be readded on next fetch unless the
TagProperty is also removed in code!
"""
getattr(instance, self.taghandler_name).remove(key=self._key, category=self._category)
[docs]class TagCategoryProperty:
"""
Tag Category Property.
"""
taghandler_name = "tags"
[docs] def __init__(self, *default_tags):
"""
Assign a property for a Tag Category, with any number of Tag keys.
This is often more useful than the `TagProperty` since it's common to want to check which
tags of a particular category the object is a member of.
Args:
*args (str or callable): Tag keys to assign to this property, using the category given
by the name of the property. Note that, if these tags are always set on the object,
if they are removed by some other means, they will be re-added when this property
is accessed. Furthermore, changing this list after the object was created, will
not remove any old tags (there is no way for the property to know if the
new list is new or not). If a callable, it will be called without arguments to
return the tag key. It is not possible to set tag `data` this way (use the
Taghandler directly for that). Tag keys are not case sensitive.
Raises:
ValueError: If the input is not a valid tag key or tuple.
Notes:
It is not possible to set Tags with a `None` category using a `TagCategoryProperty` -
use `obj.tags.add()` instead.
Example:
::
class RogueCharacter(DefaultCharacter):
guild = TagCategoryProperty("thieves_guild", "merchant_guild")
"""
self._category = ""
self._default_tags = self._parse_tag_input(*default_tags)
def _parse_tag_input(self, *args):
"""
Parse input to the property.
Args:
*args (str or callable): Tags, either as strings or `callable`, which should return
the tag key when called without arguments. Keys are not case sensitive.
Returns:
list: A list of tag keys.
"""
tags = []
for tagkey in args:
if callable(tagkey):
tagkey = tagkey()
tags.append((str(tagkey).lower()))
return tags
def __set_name__(self, cls, name):
"""
Called when descriptor is first assigned to the class (not the instance!).
It is called with the name of the field.
"""
self._category = name
def __get__(self, instance, owner):
"""
Called when accessing the tag as a property on the instance. Returns a list
of tags under the given category.
"""
taghandler = getattr(instance, self.taghandler_name)
default_tags = self._default_tags
tags = taghandler.get(category=self._category, return_list=True)
missing_default_tags = set(default_tags) - set(tags)
if missing_default_tags:
getattr(instance, self.taghandler_name).batch_add(
*[(tag, self._category) for tag in missing_default_tags]
)
tags += missing_default_tags # to avoid a second db call
return tags
def __set__(self, instance, *args):
"""
Assign a new set of tags to the category. Note that we can't know if previous
tags were assigned from this property or from TagHandler, so we don't
remove old tags. To refresh to only have the tags in this constructor, first
use `del` on this property and re-access the property with the changed default list.
"""
getattr(instance, self.taghandler_name).batch_add(*[(tag, self._category) for tag in args])
def __delete__(self, instance):
"""
Called when running `del` on the property. Will remove all tags of this
category from the object. Note that next time this desriptor is accessed, the
default ones will be re-added!
Note:
This will remove _all_ tags of this category from the object. This is necessary
in order to be able to be able to combine this with `__set__` to get a tag
list where property and handler are in sync.
"""
getattr(instance, self.taghandler_name).remove(category=self._category)
[docs]class TagHandler(object):
"""
Generic tag-handler. Accessed via TypedObject.tags.
"""
_m2m_fieldname = "db_tags"
_tagtype = None
[docs] def __init__(self, obj):
"""
Tags are stored internally in the TypedObject.db_tags m2m
field with an tag.db_model based on the obj the taghandler is
stored on and with a tagtype given by self.handlertype
Args:
obj (object): The object on which the handler is set.
"""
self.obj = obj
self._objid = obj.id
self._model = obj.__dbclass__.__name__.lower()
self._cache = {}
# store category names fully cached
self._catcache = {}
# full cache was run on all tags
self._cache_complete = False
def _query_all(self):
"""
Get all tags for this object.
"""
query = {
"%s__id" % self._model: self._objid,
"tag__db_model": self._model,
"tag__db_tagtype": self._tagtype,
}
return [
conn.tag
for conn in getattr(self.obj, self._m2m_fieldname).through.objects.filter(**query)
]
def _fullcache(self):
"""
Cache all tags of this object.
"""
if not _TYPECLASS_AGGRESSIVE_CACHE:
return
tags = self._query_all()
self._cache = dict(
(
"%s-%s"
% (
to_str(tag.db_key).lower(),
tag.db_category.lower() if tag.db_category else None,
),
tag,
)
for tag in tags
)
self._cache_complete = True
def _getcache(self, key=None, category=None):
"""
Retrieve from cache or database (always caches)
Args:
key (str, optional): Tag key to query for
category (str, optional): Tag category
Returns:
args (list): Returns a list of zero or more matches
found from cache or database.
Notes:
When given a category only, a search for all objects
of that category is done and a the category *name* is is
stored. This tells the system on subsequent calls that the
list of cached tags of this category is up-to-date
and that the cache can be queried for category matches
without missing any.
The TYPECLASS_AGGRESSIVE_CACHE=False setting will turn off
caching, causing each tag access to trigger a
database lookup.
"""
key = str(key).strip().lower() if key else None
category = category.strip().lower() if category else None
if key:
cachekey = "%s-%s" % (key, category)
tag = _TYPECLASS_AGGRESSIVE_CACHE and self._cache.get(cachekey, None)
if tag and (not hasattr(tag, "pk") and tag.pk is None):
# clear out Tags deleted from elsewhere. We must search this anew.
tag = None
del self._cache[cachekey]
if tag:
return [tag] # return cached entity
else:
query = {
"%s__id" % self._model: self._objid,
"tag__db_model": self._model,
"tag__db_tagtype": self._tagtype,
"tag__db_key__iexact": key.lower(),
"tag__db_category__iexact": category.lower() if category else None,
}
conn = getattr(self.obj, self._m2m_fieldname).through.objects.filter(**query)
if conn:
tag = conn[0].tag
if _TYPECLASS_AGGRESSIVE_CACHE:
self._cache[cachekey] = tag
return [tag]
else:
# only category given (even if it's None) - we can't
# assume the cache to be complete unless we have queried
# for this category before
catkey = "-%s" % category
if _TYPECLASS_AGGRESSIVE_CACHE and catkey in self._catcache:
return [tag for key, tag in self._cache.items() if key.endswith(catkey)]
else:
# we have to query to make this category up-date in the cache
query = {
"%s__id" % self._model: self._objid,
"tag__db_model": self._model,
"tag__db_tagtype": self._tagtype,
"tag__db_category__iexact": category.lower() if category else None,
}
tags = [
conn.tag
for conn in getattr(self.obj, self._m2m_fieldname).through.objects.filter(
**query
)
]
if _TYPECLASS_AGGRESSIVE_CACHE:
for tag in tags:
cachekey = "%s-%s" % (tag.db_key, category)
self._cache[cachekey] = tag
# mark category cache as up-to-date
self._catcache[catkey] = True
return tags
return []
def _setcache(self, key, category, tag_obj):
"""
Update cache.
Args:
key (str): A cleaned key string
category (str or None): A cleaned category name
tag_obj (tag): The newly saved tag
"""
if not _TYPECLASS_AGGRESSIVE_CACHE:
return
if not key: # don't allow an empty key in cache
return
key, category = (
str(key).strip().lower(),
category.strip().lower() if category else category,
)
cachekey = "%s-%s" % (key, category)
catkey = "-%s" % category
self._cache[cachekey] = tag_obj
# mark that the category cache is no longer up-to-date
self._catcache.pop(catkey, None)
self._cache_complete = False
def _delcache(self, key, category):
"""
Remove tag from cache
Args:
key (str): A cleaned key string
category (str or None): A cleaned category name
"""
key, category = (
str(key).strip().lower(),
category.strip().lower() if category else category,
)
catkey = "-%s" % category
if key:
cachekey = "%s-%s" % (key, category)
self._cache.pop(cachekey, None)
else:
[self._cache.pop(key, None) for key in self._cache if key.endswith(catkey)]
# mark that the category cache is no longer up-to-date
self._catcache.pop(catkey, None)
self._cache_complete = False
[docs] def reset_cache(self):
"""
Reset the cache from the outside.
"""
self._cache_complete = False
self._cache = {}
self._catcache = {}
[docs] def add(self, key=None, category=None, data=None):
"""
Add a new tag to the handler.
Args:
key (str or list): The name of the tag to add. If a list,
add several Tags.
category (str, optional): Category of Tag. `None` is the default category.
data (str, optional): Info text about the tag(s) added.
This can not be used to store object-unique info but only
eventual info about the tag itself.
Notes:
If the tag + category combination matches an already
existing Tag object, this will be re-used and no new Tag
will be created.
"""
if not key:
return
if not self._cache_complete:
self._fullcache()
for tagstr in make_iter(key):
if not tagstr:
continue
tagstr = str(tagstr).strip().lower()
category = str(category).strip().lower() if category else category
data = str(data) if data is not None else None
# this will only create tag if no matches existed beforehand (it
# will overload data on an existing tag since that is not
# considered part of making the tag unique)
tagobj = self.obj.__class__.objects.create_tag(
key=tagstr, category=category, data=data, tagtype=self._tagtype
)
getattr(self.obj, self._m2m_fieldname).add(tagobj)
self._setcache(tagstr, category, tagobj)
[docs] def has(self, key=None, category=None, return_list=False):
"""
Checks if the given Tag (or list of Tags) exists on the object.
Args:
key (str or iterable): The Tag key or tags to check for.
If `None`, search by category.
category (str, optional): Limit the check to Tags with this
category (note, that `None` is the default category).
Returns:
has_tag (bool or list): If the Tag exists on this object or not.
If `tag` was given as an iterable then the return is a list of booleans.
Raises:
ValueError: If neither `tag` nor `category` is given.
"""
ret = []
category = category.strip().lower() if category is not None else None
if key:
for tag_str in make_iter(key):
tag_str = str(tag_str).strip().lower()
ret.append(bool(self._getcache(tag_str, category)))
elif category:
ret.extend(bool(tag) for tag in self._getcache(category=category))
else:
raise ValueError("Either tag or category must be provided.")
if return_list:
return ret
return ret[0] if len(ret) == 1 else ret
[docs] def get(
self,
key=None,
default=None,
category=None,
return_tagobj=False,
return_list=False,
raise_exception=False,
):
"""
Get the tag for the given key, category or combination of the two.
Args:
key (str or list, optional): The tag or tags to retrieve.
default (any, optional): The value to return in case of no match.
category (str, optional): The Tag category to limit the
request to. Note that `None` is the valid, default
category. If no `key` is given, all tags of this category will be
returned.
return_tagobj (bool, optional): Return the Tag object itself
instead of a string representation of the Tag.
return_list (bool, optional): Always return a list, regardless
of number of matches.
raise_exception (bool, optional): Raise AttributeError if no matches
are found.
Returns:
tags (list): The matches, either string
representations of the tags or the Tag objects themselves
depending on `return_tagobj`. If 'default' is set, this
will be a list with the default value as its only element.
Raises:
AttributeError: If finding no matches and `raise_exception` is True.
"""
ret = []
for keystr in make_iter(key):
# note - the _getcache call removes case sensitivity for us
ret.extend(
[
tag if return_tagobj else to_str(tag.db_key)
for tag in self._getcache(keystr, category)
]
)
if not ret:
if raise_exception:
raise AttributeError(f"No tags found matching input {key}, {category}.")
elif return_list:
return [default] if default is not None else []
else:
return default
return ret if return_list else (ret[0] if len(ret) == 1 else ret)
[docs] def remove(self, key=None, category=None):
"""
Remove a tag from the handler based ond key and/or category.
Args:
key (str or list, optional): The tag or tags to retrieve.
category (str, optional): The Tag category to limit the
request to. Note that `None` is the valid, default
category
Notes:
If neither key nor category is specified, this acts
as .clear().
"""
if not key:
# only category
self.clear(category=category)
return
for key in make_iter(key):
if not (key or key.strip()): # we don't allow empty tags
continue
tagstr = str(key).strip().lower()
category = category.strip().lower() if category else category
# This does not delete the tag object itself. Maybe it should do
# that when no objects reference the tag anymore (but how to check)?
# For now, tags are never deleted, only their connection to objects.
tagobj = getattr(self.obj, self._m2m_fieldname).filter(
db_key=tagstr, db_category=category, db_model=self._model, db_tagtype=self._tagtype
)
if tagobj:
getattr(self.obj, self._m2m_fieldname).remove(tagobj[0])
self._delcache(key, category)
[docs] def clear(self, category=None):
"""
Remove all tags from the handler.
Args:
category (str, optional): The Tag category to limit the
request to. Note that `None` is the valid, default
category.
"""
if not self._cache_complete:
self._fullcache()
query = {
"%s__id" % self._model: self._objid,
"tag__db_model": self._model,
"tag__db_tagtype": self._tagtype,
}
if category:
query["tag__db_category"] = category.strip().lower()
getattr(self.obj, self._m2m_fieldname).through.objects.filter(**query).delete()
self._cache = {}
self._catcache = {}
self._cache_complete = False
[docs] def all(self, return_key_and_category=False, return_objs=False):
"""
Get all tags in this handler, regardless of category.
Args:
return_key_and_category (bool, optional): Return a list of
tuples `[(key, category), ...]`.
return_objs (bool, optional): Return tag objects.
Returns:
tags (list): A list of tag keys `[tagkey, tagkey, ...]` or
a list of tuples `[(key, category), ...]` if
`return_key_and_category` is set.
"""
if _TYPECLASS_AGGRESSIVE_CACHE:
if not self._cache_complete:
self._fullcache()
tags = sorted(self._cache.values())
else:
tags = sorted(self._query_all())
if return_key_and_category:
# return tuple (key, category)
return [(to_str(tag.db_key), tag.db_category) for tag in tags]
elif return_objs:
return tags
else:
return [to_str(tag.db_key) for tag in tags]
[docs] def batch_add(self, *args):
"""
Batch-add tags from a list of tuples.
Args:
*args (tuple or str): Each argument should be a `tagstr` keys or tuple
`(keystr, category)` or `(keystr, category, data)`. It's possible to mix input
types.
Notes:
This will generate a mimimal number of self.add calls,
based on the number of categories involved (including
`None`) (data is not unique and may be overwritten by the content
of a latter tuple with the same category).
"""
keys = defaultdict(list)
data = {}
for tup in args:
tup = make_iter(tup)
nlen = len(tup)
if nlen == 1: # just a key, no category
keys[None].append(tup[0])
elif nlen == 2:
keys[tup[1]].append(tup[0])
else:
keys[tup[1]].append(tup[0])
data[tup[1]] = tup[2] # overwrite previous
for category, key in keys.items():
self.add(key=key, category=category, data=data.get(category, None))
[docs] def batch_remove(self, *args):
"""
Batch-remove tags from a list of tuples.
Args:
*args (tuple or str): Each argument should be a `tagstr` keys or tuple
`(keystr, category)` or `(keystr, category, data)` (the `data` field is ignored,
only `keystr`/`category` matters). It's possible to mix input types.
"""
keys = defaultdict(list)
for tup in args:
tup = make_iter(tup)
nlen = len(tup)
if nlen == 1: # just a key, no category
keys[None].append(tup[0])
elif nlen > 1:
keys[tup[1]].append(tup[0])
for category, key in keys.items():
self.remove(key=key, category=category, data=data.get(category, None))
def __str__(self):
return ",".join(self.all())
[docs]class AliasProperty(TagProperty):
"""
Allows for setting aliases like Django fields:
::
class Character(DefaultCharacter):
# note that every character will get the alias bob. Make sure
# the alias property does not collide with an existing method
# or property on the class.
bob = AliasProperty()
"""
taghandler_name = "aliases"
[docs]class AliasHandler(TagHandler):
"""
A handler for the Alias Tag type.
"""
_tagtype = "alias"
[docs]class PermissionProperty(TagProperty):
"""
Allows for setting permissions like Django fields:
::
class Character(DefaultCharacter):
# note that every character will get this permission! Make
# sure it doesn't collide with an existing method or property.
myperm = PermissionProperty()
"""
taghandler_name = "permissions"
[docs]class PermissionHandler(TagHandler):
"""
A handler for the Permission Tag type.
"""
_tagtype = "permission"
[docs] def check(self, *permissions, require_all=False):
"""
Straight-up check the provided permission against this handler. The check will pass if
- any/all given permission exists on the handler (depending on if `require_all` is set).
- If handler sits on puppeted object and this is a hierarachical perm, the puppeting
Account's permission will also be included in the check, prioritizing the Account's perm
(this avoids escalation exploits by puppeting a too-high prio character)
- a permission is also considered to exist on the handler, if it is *lower* than
a permission on the handler and this is a 'hierarchical' permission given
in `settings.PERMISSION_HIERARCHY`. Example: If the 'Developer' hierarchical
perm perm is set on the handler, and we check for the 'Builder' perm, the
check will pass.
Args:
*permissions (str): Any number of permissions to check. By default,
the permission is passed if any of these (or higher, if a
hierarchical permission defined in settings.PERMISSION_HIERARCHY)
exists in the handler. Permissions are not case-sensitive.
require_all (bool): If set, *all* provided permissions much pass
the check for the entire check to pass. By default only one
needs to pass.
Returns:
bool: If the provided permission(s) pass the check on this handler.
Example:
::
can_enter = obj.permissions.check("Blacksmith", "Builder")
Notes:
This works the same way as the `perms` lockfunc and could be
replicated with a lock check against the lockstring
"locktype: perm(perm1) OR perm(perm2) OR ..."
(using AND for the `require_all` condition).
"""
if require_all:
return all(perm_lockfunc(self.obj, None, perm) for perm in permissions)
else:
return any(perm_lockfunc(self.obj, None, perm) for perm in permissions)
