# pylint: disable=invalid-name, no-member
'''Database model additions for Kata'''
import datetime
import logging
import sqlalchemy.orm as orm
from sqlalchemy.schema import Table, Column, UniqueConstraint, ForeignKey
import sqlalchemy.types as types
import vdm.sqlalchemy
import ckan.model as model
from ckan.model.domain_object import DomainObject
import ckan.model.domain_object as domain_object
from ckan.model import meta, extension, user
import ckan.model.types as _types
mapper = orm.mapper
log = logging.getLogger(__name__)
[docs]class KataAccessRequest(DomainObject):
'''
Class for access requests.
'''
def __init__(self, follower_id, object_id):
self.user_id = follower_id
self.pkg_id = object_id
@classmethod
[docs] def get(cls, follower_id, object_id):
'''
Return a `UserFollowingDataset` object for the given `follower_id` and
`object_id`, or None if no such follower exists.
'''
query = meta.Session.query(KataAccessRequest)
query = query.filter(KataAccessRequest.user_id == follower_id)
query = query.filter(KataAccessRequest.pkg_id == object_id)
return query.first()
@classmethod
[docs] def is_requesting(cls, follower_id, object_id):
'''
Return `True` if `follower_id` is currently following `object_id`, `False`
otherwise.
'''
return KataAccessRequest.get(follower_id, object_id) is not None
def _create_extra(key, value):
'''
Create a UserExtra instance.
'''
return UserExtra(key=unicode(key), value=value)
[docs]def setup():
'''
Creates the tables that are specified in this file
'''
if model.package_table.exists() and not kata_access_request_table.exists():
kata_access_request_table.create()
log.debug('Kata access request table created')
if model.user_table.exists() and not user_extra_table.exists():
user_extra_table.create()
log.debug('User extra table created')
[docs]def delete_tables():
'''
Delete data from some extra tables to prevent IntegrityError between tests.
'''
#if user_extra_table.exists():
#user_extra_table.delete()
if kata_access_request_table.exists():
kata_access_request_table.delete()
kata_access_request_table = Table('kata_req', meta.metadata,
Column('id', types.UnicodeText, primary_key=True, default=_types.make_uuid),
Column('pkg_id', types.UnicodeText, nullable=False),
Column('user_id', types.UnicodeText, nullable=False),
Column('created', types.DateTime, default=datetime.datetime.utcnow),
UniqueConstraint('pkg_id', 'user_id', name='pkgusr_1'),
)
mapper(KataAccessRequest, kata_access_request_table, extension=[
extension.PluginMapperExtension(),
])
user_extra_table = Table('user_extra', meta.metadata,
Column('id', types.UnicodeText, primary_key=True, default=_types.make_uuid),
Column('user_id', types.UnicodeText, ForeignKey('user.id')),
Column('key', types.UnicodeText),
Column('value', _types.JsonType),
)
vdm.sqlalchemy.make_table_stateful(user_extra_table)
meta.mapper(UserExtra, user_extra_table, properties={
'user': orm.relation(user.User,
backref=orm.backref('_extras',
collection_class=orm.collections.attribute_mapped_collection(u'key'),
cascade='all, delete, delete-orphan',
),
)
},
order_by=[user_extra_table.c.user_id, user_extra_table.c.key],
)
extras_active = vdm.sqlalchemy.stateful.DeferredProperty('_extras',
vdm.sqlalchemy.stateful.StatefulDict,
base_modifier=lambda x: x.get_as_of())
setattr(user.User, 'extras_active', extras_active)
user.User.extras = vdm.sqlalchemy.stateful.OurAssociationProxy('extras_active', 'value', creator=_create_extra)