[wpp.git] / app / tools.py

from app import db
from .models import User, Portfolio, Album, Photo, AccessRight, Role, AlbumAcl, PortfolioAcl

class DBQueryException(Exception):
    pass
    
class DBQuery:
    
    def get_albums(self, object):
        albums = []
        if hasattr(object, '__tablename__'):
            if isinstance(object, User):
                portfolios = Portfolio.query.filter_by(owner=object)
                for p in portfolios:
                    albums.extend(Album.query.filter_by(portfolio=p))
            elif isinstance(object, Portfolio):
                albums = Album.query.filter_by(portfolio=object)
            elif isinstance(object, Photo):
                albums.append(object.album)
            else:
                raise DBQueryException("%s: not supported" % object.__tablename__)
            result = []
            for album in albums:
                result.append(album)
            return result
        else:
            raise DBQueryException("{}: not supported".format(object))

    def get_portfolios(self, object):
        portfolios = []
        if hasattr(object, '__tablename__'):
            if isinstance(object, User):
                portfolios = Portfolio.query.filter_by(owner=object)
            elif isinstance(object, Album):
                portfolios.append(object.portfolio)
            elif isinstance(object, Photo):
                album = self.get_albums(object)
                portfolios.append(album[0].portfolio)
            else:
                raise DBQueryException("%s: not supported" % object.__tablename__)
            result = []
            for portfolio in portfolios:
                result.append(portfolio)
            return result
        else:
            raise DBQueryException("{}: not supported".format(object))

    def get_owner(self, object):
        if hasattr(object, '__tablename__'):
            if isinstance(object, Portfolio):
                return object.owner
            elif isinstance(object, Album):
                p = object.portfolio
                return p.owner
            elif isinstance(object, Photo):
                a = object.album
                p = a.portfolio
                return p.owner
            else:
                raise DBQueryException("%s: not supported" % object.__tablename__)
        else:
            raise DBQueryException("{}: not supported".format(object))

    def get_users(self, object):
        if hasattr(object, '__tablename__'):
            users = []
            if isinstance(object, Album):
                objects = AccessRight.query.join(AccessRight.album).filter(Album.id == object.id)
                for o in objects:
                    users.append(o.user)
            elif isinstance(object, Photo):
                objects = AccessRight.query.join(AccessRight.photo).filter(Photo.id == object.id)
                for o in objects:
                    users.append(o.user)
            elif isinstance(object, Portfolio):
                objects = AccessRight.query.join(AccessRight.portfolio).filter(Portfolio.id == object.id)
                for o in objects:
                    users.append(o.user)
            else:
                raise DBQueryException("%s: not supported" % object.__tablename__)
            return users
        else:
            raise DBQueryException("{}: not supported".format(object))

    def has_access(self, user, object):
        if hasattr(user, '__tablename__') and hasattr(object, '__tablename__') and isinstance(user, User):
            result = False
            if isinstance(object, Portfolio):
                acl = self.get_acl(user, object)
                if acl is not None:
                    result = True
            elif isinstance(object, Album):
                acl = self.get_acl(user, object)
                if acl is not None:
                    result = True
                else:
                    portfolio = self.get_portfolios(object)
                    acl = self.get_acl(user, portfolio[0])
                    if acl is not None:
                        result = True
            elif isinstance(object, Photo):
                acl = self.get_acl(user, object)
                if acl is not None:
                    result = True
                else:
                    albums = self.get_albums(object)
                    acl = self.get_acl(user, albums[0])
                    if acl is not None:
                        result = True
                    else:
                        portfolio = self.get_portfolios(object)
                        acl = self.get_acl(user, portfolio[0])
                        if acl is not None:
                            result = True
            else:
                raise DBQueryException("%s: not supported" % object.__tablename__)
            return result
        else:
            raise DBQueryException("{}: not supported".format(object))
        
    def can_read(self, user, object):
        return self.has_access(user, object)

    def can_write(self, user, object):
        if hasattr(user, '__tablename__') and hasattr(object, '__tablename__') and isinstance(user, User):
            result = False
            if isinstance(object, Portfolio):
                acl = self.get_acl(user, object)
                if acl is not None and acl > Role.read:
                    result = True
            elif isinstance(object, Album):
                acl = self.get_acl(user, object)
                if acl is not None and acl > Role.read:
                    result = True
                else:
                    portfolio = self.get_portfolios(object)
                    acl = self.get_acl(user, portfolio[0])
                    if acl is not None and acl > Role.read:
                        result = True
            elif isinstance(object, Photo):
                acl = self.get_acl(user, object)
                if acl is not None and acl > Role.read:
                    result = True
                else:
                    albums = self.get_albums(object)
                    acl = self.get_acl(user, albums[0])
                    if acl is not None and acl > Role.read:
                        result = True
                    else:
                        portfolio = self.get_portfolios(object)
                        acl = self.get_acl(user, portfolio[0])
                        if acl is not None and acl > Role.read:
                            result = True
            else:
                raise DBQueryException("%s: not supported" % object.__tablename__)
            return result
        else:
            raise DBQueryException("{}: not supported".format(object))

    def get_acl(self, user, object): 
        if hasattr(user, '__tablename__') and hasattr(object, '__tablename__') and isinstance(user, User):
            if isinstance(object, Album):
                if user.is_admin or user == self.get_owner(object):
                    return Role.admin
                acl = AccessRight.query.join(AccessRight.album).filter(Album.id == object.id)
                for a in acl:
                    if AccessRight.query.get(a.id).user_id == user.id:
                        return a.right
                # Check if access given by parent (Portfolio)
                return self.get_acl(user, object.portfolio)
            elif isinstance(object, Photo):
                if user.is_admin or user == self.get_owner(object):
                    return Role.admin
                acl = AccessRight.query.join(AccessRight.photo).filter(Photo.id == object.id)
                for a in acl:
                    if AccessRight.query.get(a.id).user_id == user.id:
                        return a.right
                # Check if access given by parent (Album)
                return self.get_acl(user, object.album)
            elif isinstance(object, Portfolio):
                if user.is_admin or user == self.get_owner(object):
                    return Role.admin
                acl = AccessRight.query.join(AccessRight.portfolio).filter(Portfolio.id == object.id)
                for a in acl:
                    if AccessRight.query.get(a.id).user_id == user.id:
                        return a.right
            else:
                raise DBQueryException("%s: not supported" % object.__tablename__)
            return None
        else:
            raise DBQueryException("{}: not supported".format(object))

    def get_albums_for_user(self, user, owner = False, hidden = False):
        if isinstance(user, User):
            albums = []
            # Find all albums with direct access
            objects = db.session.query(AlbumAcl, Album, AccessRight, Portfolio).join(
                Album, AlbumAcl.c.album == Album.id).join(
                AccessRight, AccessRight.id == AlbumAcl.c.acl).join(
                Portfolio, Portfolio.id == Album.portfolio_id).filter(
                AccessRight.user == user).all()
            for object in objects:
                if not owner and object.Portfolio.owner == user:
                    continue
                if not hidden and not object.Album.visible:
                    continue
                albums.append(object.Album)
            # Find all albums with access through inherited access rights from portfolio
            temp = []
            objects = db.session.query(PortfolioAcl, Album, AccessRight, Portfolio).join(
                Portfolio, PortfolioAcl.c.portfolio == Portfolio.id).join(
                AccessRight, AccessRight.id == PortfolioAcl.c.acl).join(
                Album, Album.portfolio_id == Portfolio.id).all()
            for object in objects:
                if not owner and object.Portfolio.owner == user:
                    continue
                if not hidden and not object.Album.visible:
                    continue
                temp.append(object.Album)
            # Merge albums with temp removing duplicates
            albums = albums + list(set(temp) - set(albums))
            return albums
        else:
            raise DBQueryException("%s: not supported" % object.__tablename__)
Commit	Line	Data
b1f72147 MR	1	from app import db
	2	from .models import User, Portfolio, Album, Photo, AccessRight, Role, AlbumAcl, PortfolioAcl
	3
	4	class DBQueryException(Exception):
	5	pass
	6
	7	class DBQuery:
	8
	9	def get_albums(self, object):
	10	albums = []
	11	if hasattr(object, '__tablename__'):
	12	if isinstance(object, User):
	13	portfolios = Portfolio.query.filter_by(owner=object)
	14	for p in portfolios:
	15	albums.extend(Album.query.filter_by(portfolio=p))
	16	elif isinstance(object, Portfolio):
	17	albums = Album.query.filter_by(portfolio=object)
	18	elif isinstance(object, Photo):
	19	albums.append(object.album)
	20	else:
	21	raise DBQueryException("%s: not supported" % object.__tablename__)
	22	result = []
	23	for album in albums:
	24	result.append(album)
	25	return result
	26	else:
	27	raise DBQueryException("{}: not supported".format(object))
	28
	29	def get_portfolios(self, object):
	30	portfolios = []
	31	if hasattr(object, '__tablename__'):
	32	if isinstance(object, User):
	33	portfolios = Portfolio.query.filter_by(owner=object)
	34	elif isinstance(object, Album):
	35	portfolios.append(object.portfolio)
	36	elif isinstance(object, Photo):
	37	album = self.get_albums(object)
	38	portfolios.append(album[0].portfolio)
	39	else:
	40	raise DBQueryException("%s: not supported" % object.__tablename__)
	41	result = []
	42	for portfolio in portfolios:
	43	result.append(portfolio)
	44	return result
	45	else:
	46	raise DBQueryException("{}: not supported".format(object))
	47
	48	def get_owner(self, object):
	49	if hasattr(object, '__tablename__'):
	50	if isinstance(object, Portfolio):
	51	return object.owner
	52	elif isinstance(object, Album):
	53	p = object.portfolio
	54	return p.owner
	55	elif isinstance(object, Photo):
	56	a = object.album
	57	p = a.portfolio
	58	return p.owner
	59	else:
	60	raise DBQueryException("%s: not supported" % object.__tablename__)
	61	else:
	62	raise DBQueryException("{}: not supported".format(object))
	63
	64	def get_users(self, object):
65	if hasattr(object, '__tablename__'):
66	users = []
67	if isinstance(object, Album):
68	objects = AccessRight.query.join(AccessRight.album).filter(Album.id == object.id)
69	for o in objects:
70	users.append(o.user)
71	elif isinstance(object, Photo):
72	objects = AccessRight.query.join(AccessRight.photo).filter(Photo.id == object.id)
73	for o in objects:
74	users.append(o.user)
75	elif isinstance(object, Portfolio):
76	objects = AccessRight.query.join(AccessRight.portfolio).filter(Portfolio.id == object.id)
77	for o in objects:
78	users.append(o.user)
79	else:
80	raise DBQueryException("%s: not supported" % object.__tablename__)
81	return users
82	else:
83	raise DBQueryException("{}: not supported".format(object))
84
85	def has_access(self, user, object):
86	if hasattr(user, '__tablename__') and hasattr(object, '__tablename__') and isinstance(user, User):
87	result = False
88	if isinstance(object, Portfolio):
89	acl = self.get_acl(user, object)
90	if acl is not None:
91	result = True
92	elif isinstance(object, Album):
93	acl = self.get_acl(user, object)
94	if acl is not None:
95	result = True
96	else:
97	portfolio = self.get_portfolios(object)
98	acl = self.get_acl(user, portfolio[0])
99	if acl is not None:
100	result = True
101	elif isinstance(object, Photo):
102	acl = self.get_acl(user, object)
103	if acl is not None:
104	result = True
105	else:
106	albums = self.get_albums(object)
107	acl = self.get_acl(user, albums[0])
108	if acl is not None:
109	result = True
110	else:
111	portfolio = self.get_portfolios(object)
112	acl = self.get_acl(user, portfolio[0])
113	if acl is not None:
114	result = True
115	else:
116	raise DBQueryException("%s: not supported" % object.__tablename__)
117	return result
118	else:
119	raise DBQueryException("{}: not supported".format(object))
120
121	def can_read(self, user, object):
122	return self.has_access(user, object)
123
124	def can_write(self, user, object):
125	if hasattr(user, '__tablename__') and hasattr(object, '__tablename__') and isinstance(user, User):
126	result = False
127	if isinstance(object, Portfolio):
128	acl = self.get_acl(user, object)
129	if acl is not None and acl > Role.read:
130	result = True
131	elif isinstance(object, Album):
132	acl = self.get_acl(user, object)
133	if acl is not None and acl > Role.read:
134	result = True
135	else:
136	portfolio = self.get_portfolios(object)
137	acl = self.get_acl(user, portfolio[0])
138	if acl is not None and acl > Role.read:
139	result = True
140	elif isinstance(object, Photo):
141	acl = self.get_acl(user, object)
142	if acl is not None and acl > Role.read:
143	result = True
144	else:
145	albums = self.get_albums(object)
146	acl = self.get_acl(user, albums[0])
147	if acl is not None and acl > Role.read:
148	result = True
149	else:
150	portfolio = self.get_portfolios(object)
151	acl = self.get_acl(user, portfolio[0])
152	if acl is not None and acl > Role.read:
153	result = True
154	else:
155	raise DBQueryException("%s: not supported" % object.__tablename__)
156	return result
157	else:
158	raise DBQueryException("{}: not supported".format(object))
159
160	def get_acl(self, user, object):
161	if hasattr(user, '__tablename__') and hasattr(object, '__tablename__') and isinstance(user, User):
162	if isinstance(object, Album):
163	if user.is_admin or user == self.get_owner(object):
164	return Role.admin
165	acl = AccessRight.query.join(AccessRight.album).filter(Album.id == object.id)
166	for a in acl:
167	if AccessRight.query.get(a.id).user_id == user.id:
168	return a.right
169	# Check if access given by parent (Portfolio)
170	return self.get_acl(user, object.portfolio)
171	elif isinstance(object, Photo):
172	if user.is_admin or user == self.get_owner(object):
173	return Role.admin
174	acl = AccessRight.query.join(AccessRight.photo).filter(Photo.id == object.id)
175	for a in acl:
176	if AccessRight.query.get(a.id).user_id == user.id:
177	return a.right
178	# Check if access given by parent (Album)
179	return self.get_acl(user, object.album)
180	elif isinstance(object, Portfolio):
181	if user.is_admin or user == self.get_owner(object):
182	return Role.admin
183	acl = AccessRight.query.join(AccessRight.portfolio).filter(Portfolio.id == object.id)
184	for a in acl:
185	if AccessRight.query.get(a.id).user_id == user.id:
186	return a.right
187	else:
188	raise DBQueryException("%s: not supported" % object.__tablename__)
189	return None
190	else:
191	raise DBQueryException("{}: not supported".format(object))
192
193	def get_albums_for_user(self, user, owner = False, hidden = False):
194	if isinstance(user, User):
195	albums = []
196	# Find all albums with direct access
197	objects = db.session.query(AlbumAcl, Album, AccessRight, Portfolio).join(
198	Album, AlbumAcl.c.album == Album.id).join(
199	AccessRight, AccessRight.id == AlbumAcl.c.acl).join(
200	Portfolio, Portfolio.id == Album.portfolio_id).filter(
201	AccessRight.user == user).all()
202	for object in objects:
203	if not owner and object.Portfolio.owner == user:
204	continue
205	if not hidden and not object.Album.visible:
206	continue
207	albums.append(object.Album)
208	# Find all albums with access through inherited access rights from portfolio
209	temp = []
210	objects = db.session.query(PortfolioAcl, Album, AccessRight, Portfolio).join(
211	Portfolio, PortfolioAcl.c.portfolio == Portfolio.id).join(
212	AccessRight, AccessRight.id == PortfolioAcl.c.acl).join(
213	Album, Album.portfolio_id == Portfolio.id).all()
214	for object in objects:
215	if not owner and object.Portfolio.owner == user:
216	continue
217	if not hidden and not object.Album.visible:
218	continue
219	temp.append(object.Album)
220	# Merge albums with temp removing duplicates
221	albums = albums + list(set(temp) - set(albums))
222	return albums
223	else:
224	raise DBQueryException("%s: not supported" % object.__tablename__)