115 lines
3.4 KiB
Python
115 lines
3.4 KiB
Python
|
|
import collections
|
|
import json
|
|
|
|
from datetime import datetime
|
|
|
|
from sqlalchemy import Column, Integer, String, DateTime
|
|
from sqlalchemy import ForeignKey
|
|
from sqlalchemy.ext.declarative import declarative_base
|
|
from sqlalchemy.ext.mutable import Mutable
|
|
from sqlalchemy.orm import column_property, relationship, backref
|
|
from sqlalchemy.types import TypeDecorator, VARCHAR
|
|
|
|
|
|
class JSONEncodedDict(TypeDecorator):
|
|
"Represents an immutable structure as a json-encoded string."
|
|
|
|
impl = VARCHAR
|
|
|
|
def process_bind_param(self, value, dialect):
|
|
if value is not None:
|
|
value = json.dumps(value)
|
|
return value
|
|
|
|
def process_result_value(self, value, dialect):
|
|
if value is not None:
|
|
value = json.loads(value)
|
|
return value
|
|
|
|
class MutationDict(Mutable, dict):
|
|
|
|
@classmethod
|
|
def coerce(cls, key, value):
|
|
""" Convert plain dictionaries to MutationDict. """
|
|
if not isinstance(value, MutationDict):
|
|
if isinstance(value, dict):
|
|
return MutationDict(value)
|
|
# this call will raise ValueError
|
|
return Mutable.coerce(key, value)
|
|
else:
|
|
return value
|
|
|
|
def __setitem__(self, key, value):
|
|
""" Detect dictionary set events and emit change events. """
|
|
dict.__setitem__(self, key, value)
|
|
self.changed()
|
|
|
|
def __delitem__(self, key):
|
|
""" Detect dictionary del events and emit change events. """
|
|
dict.__delitem__(self, key)
|
|
self.changed()
|
|
|
|
MutationDict.associate_with(JSONEncodedDict)
|
|
Base = declarative_base()
|
|
|
|
class Site(Base):
|
|
__tablename__ = 'sites'
|
|
|
|
id = Column(Integer, primary_key=True)
|
|
name = Column(String(50), unique=True, nullable=False)
|
|
|
|
def __init__(self, name):
|
|
self.name = name
|
|
|
|
def __repr__(self):
|
|
return "<Site('%s')>" % (self.name)
|
|
|
|
class LocalUser(Base):
|
|
""" Mandaye's user
|
|
"""
|
|
__tablename__ = 'local_users'
|
|
|
|
id = Column(Integer, primary_key=True)
|
|
login = Column(String(150), nullable=False, unique=True)
|
|
password = Column(String(25), nullable=True)
|
|
firstname = Column(String(150), nullable=True)
|
|
lastname = Column(String(150), nullable=True)
|
|
fullname = column_property(firstname + " " + lastname)
|
|
|
|
creation_date = Column(DateTime, default=datetime.now(), nullable=False)
|
|
last_connection = Column(DateTime, default=datetime.now())
|
|
|
|
def __init__(self, login=None, password=None, fullname=None):
|
|
self.login = login
|
|
self.password = password
|
|
self.fullname = fullname
|
|
|
|
def __repr__(self):
|
|
return "<LocalUser('%d %s')>" % (self.id, self.fullname)
|
|
|
|
class ExtUser(Base):
|
|
""" User of externals applications
|
|
"""
|
|
__tablename__ = 'ext_users'
|
|
|
|
id = Column(Integer, primary_key=True)
|
|
login = Column(String(150), nullable=False)
|
|
post_values = Column(JSONEncodedDict)
|
|
creation_date = Column(DateTime, default=datetime.now(), nullable=False)
|
|
last_connection = Column(DateTime, default=datetime.now())
|
|
|
|
local_user_id = Column(Integer, ForeignKey('local_users.id'), nullable=False)
|
|
site_id = Column(Integer, ForeignKey('sites.id'), nullable=False)
|
|
local_user = relationship("LocalUser", backref=backref('ext_users'))
|
|
site = relationship("Site", backref=backref('users'))
|
|
|
|
def __init__(self, login=None, post_values=None):
|
|
self.login = login
|
|
self.post_values = post_values
|
|
|
|
def __repr__(self):
|
|
return "<ExtUser '%d'>" % (self.id)
|
|
|
|
|