Source code for alkali.model

from collections import OrderedDict
from import Iterable
import json

from .memoized_property import memoized_property
from .metamodel import MetaModel
from .utils import tznow
from . import fields
from . import signals

[docs]class ObjectDoesNotExist(Exception): """ base class for a model specific exception (eg. MyModel.DoesNotExist) raised when a query yields no results """ pass
[docs]class Model(metaclass=MetaModel): """ main class for the database. the definition of this class defines a table schema but instances of this class hold a row. model fields are available as attributes. eg. ``m.my_field = 'foo'`` the Django docs at will be fairly relevant to alkali see :mod:`alkali.database` for some example code """ def __init__(self, *args, **kw): # MetaModel.__call__ has put fields in self, # put any other keywords into self for name, value in kw.items(): setattr(self, name, value) # note, this is called twice, once during initial object creation # and then again when this object is copied during a save signals.creation.send(self.__class__, instance=self) # called via copy.copy() module, when getting from manager def __copy__(self): new = type(self)() new.__dict__.update(self.__dict__) return new def __repr__(self): return "<{}: {}>".format(self.__class__.__name__, def __eq__(self, other): # this is obviously a very shallow comparison return type(self) == type(other) and == def __ne__(self, other): return not self.__eq__(other)
[docs] def set_field(self, field, value): """ set a field value, this method is automatically called when setting a field value. safe to call externally. fires :data:`alkali.signals.field_update` for any listeners :param field: instance of Field :type field: :class:`alkali.fields.Field` :param value: the already-cast value to store :type value: ``Field.field_type`` """ # if we're setting a field value and that value is different # than current value, mark self as modified curr_val = getattr(self, # do not let user change pk after it has been set if field.primary_key: if curr_val is not None and curr_val != value: _vals = (self.__class__.__name__,, value) raise RuntimeError( "{}: trying to change set pk value: {} to {}".format(*_vals) ) self.__dict__[] = value # actually set the value if curr_val != value: self.__dict__['_dirty'] = True signals.field_update.send(self.__class__,, old_val=curr_val, new_val=value) # call any auto fields on this model if self.__dict__['_dirty']: for name, field_class in self.Meta.fields.items(): if getattr(field_class, 'auto_now', False): self.__dict__[name] = tznow()
@property def dirty(self): """ **property**: return True if our fields have changed since creation """ return self._dirty @property def schema(self): """ **property**: a string that quickly shows the fields and types """ def fmt(name, field): return "{}:{}".format(name, field.field_type.__name__) name_type = [ fmt(n, f) for n, f in self.Meta.fields.items() ] fields = ", ".join( name_type ) return "<{}: {}>".format(self.__class__.__name__, fields) @memoized_property def pk(self): """ **property**: returns this models primary key value. If the model is comprised of several primary keys then return a tuple of them. :rtype: ``Field.field_type`` or tuple-of-Field.field_type """ pks = list(self.Meta.pk_fields.values()) foreign_pks = list(filter(lambda f: isinstance(f, fields.ForeignKey), pks)) if foreign_pks: pk_vals = tuple( getattr(self, for f in pks ) if len(pk_vals) == 1: return pk_vals[0] assert False, "not actually supported at this time" # pragma: nocover return pk_vals # pragma: nocover, not actually supported at this time else: pk_vals = tuple( getattr(self, for f in pks ) if len(pk_vals) == 1: return pk_vals[0] return pk_vals @property def valid_pk(self): pk = if isinstance(pk, Iterable): return all( map(lambda e: e is not None, pk) ) return pk is not None @property def dict(self): """ **property**: returns a dict of all the fields, the fields are json consumable :rtype: ``OrderedDict`` """ return OrderedDict( [(name, field.dumps(getattr(self, name))) for name, field in self.Meta.fields.items() ]) @property def json(self): """ **property**: returns json that holds all the fields :rtype: ``str`` """ return json.dumps(self.dict)
[docs] def save(self): """ add ourselves to our :class:`alkali.manager.Manager` and mark ourselves as no longer dirty. it's up to our ``Manager`` to persistently save us """ self._dirty = False return self