Source

djsqlalchemy / djsqlalchemy / managers.py

Full commit
from django.db import models

from djsqlalchemy.alchemy import get_tables
from djsqlalchemy.mapper import get_list, get_one, get_values
from djsqlalchemy.utils import cached_property


class DjSqlAlchemyManager(object):
    def __init__(self, model, *args, **kwargs):
        self.model = model

    @cached_property
    def tables(self):
        return get_tables()

    def get_select(self):
        raise NotImplementedError()

    def values_list(self, columns=None, flat=False):
        if hasattr(self, 'result'):
            return self.result
        else:
            return list(get_values(self.get_select(), columns=columns))

    def all(self):
        if hasattr(self, 'result'):
            return self.result
        else:
            return get_list(self.get_select(), self.model)

    def get(self):
        if hasattr(self, 'result'):
            if not self.result:
                raise self.model.DoesNotExist()
            elif len(self.result) > 1:
                raise self.model.MultipleObjectsReturned()
            else:
                return self.result.pop()
        else:
            return get_one(self.get_select(), self.model)

    def count(self):
        if hasattr(self, 'result'):
            return len(self.result)
        else:
            s = self.get_select().alias('row').count()
            return get_one(s)[0]

    def get_query_set(self):
        raise NotImplementedError('Not supported')


class ManagerPlaceholder(models.Manager):
    def __init__(self, manager):
        super(ManagerPlaceholder, self).__init__()
        self.manager = manager

    def contribute_to_class(self, model, name):
        self.model = model
        setattr(model, name, self)

    def __get__(self, instance, owner):
        if instance != None:
            raise AttributeError("Manager isn't accessible via %s instances" %
                                 type.__name__)
        return self.manager(self.model)