git.datanom.net - wpp.git/blob

1 from app import db

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

 class DBQueryException(Exception):

5 pass

7 class DBQuery:

     def get_albums(self, object):

10 albums = []

         if hasattr(object, '__tablename__'):

             if isinstance(object, User):

                 portfolios = Portfolio.query.filter_by(owner=object)

14 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)

20 else:

                 raise DBQueryException("%s: not supported" % object.__tablename__)

22 result = []

23 for album in albums:

24 result.append(album)

25 return result

26 else:

             raise DBQueryException("{}: not supported".format(object))

     def get_portfolios(self, object):

30 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)

39 else:

                 raise DBQueryException("%s: not supported" % object.__tablename__)

41 result = []

42 for portfolio in portfolios:

43 result.append(portfolio)

44 return result

45 else:

             raise DBQueryException("{}: not supported".format(object))

     def get_owner(self, object):

         if hasattr(object, '__tablename__'):

             if isinstance(object, Portfolio):

51 return object.owner

             elif isinstance(object, Album):

53 p = object.portfolio

54 return p.owner

             elif isinstance(object, Photo):

56 a = object.album

57 p = a.portfolio

58 return p.owner

59 else:

                 raise DBQueryException("%s: not supported" % object.__tablename__)

61 else:

             raise DBQueryException("{}: not supported".format(object))

     def get_users(self, object):

         if hasattr(object, '__tablename__'):

66 users = []

             if isinstance(object, Album):

                 objects = AccessRight.query.join(AccessRight.album).filter(Album.id == object.id)

69 for o in objects:

                     users.append(o.user)

             elif isinstance(object, Photo):

                 objects = AccessRight.query.join(AccessRight.photo).filter(Photo.id == object.id)

73 for o in objects:

                     users.append(o.user)

             elif isinstance(object, Portfolio):

                 objects = AccessRight.query.join(AccessRight.portfolio).filter(Portfolio.id == object.id)

77 for o in objects:

                     users.append(o.user)

79 else:

                 raise DBQueryException("%s: not supported" % object.__tablename__)

81 return users

82 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):

87 result = False

             if isinstance(object, Portfolio):

                 acl = self.get_acl(user, object)

90 if acl is not None:

91 result = True

             elif isinstance(object, Album):

                 acl = self.get_acl(user, object)

94 if acl is not None:

95 result = True

96 else:

                     portfolio = self.get_portfolios(object)

                     acl = self.get_acl(user, portfolio[0])

99 if acl is not None:

100 result = True

             elif isinstance(object, Photo):

                 acl = self.get_acl(user, object)

103 if acl is not None:

104 result = True

105 else:

                     albums = self.get_albums(object)

                     acl = self.get_acl(user, albums[0])

108 if acl is not None:

109 result = True

110 else:

                         portfolio = self.get_portfolios(object)

                         acl = self.get_acl(user, portfolio[0])

113 if acl is not None:

114 result = True

115 else:

                 raise DBQueryException("%s: not supported" % object.__tablename__)

117 return result

118 else:

             raise DBQueryException("{}: not supported".format(object))

120

     def can_read(self, user, object):

         return self.has_access(user, object)

123

     def can_write(self, user, object):

         if hasattr(user, '__tablename__') and hasattr(object, '__tablename__') and isinstance(user, User):

126 result = False

             if isinstance(object, Portfolio):

                 acl = self.get_acl(user, object)

                 if acl is not None and acl > Role.read:

130 result = True

             elif isinstance(object, Album):

                 acl = self.get_acl(user, object)

                 if acl is not None and acl > Role.read:

134 result = True

135 else:

                     portfolio = self.get_portfolios(object)

                     acl = self.get_acl(user, portfolio[0])

                     if acl is not None and acl > Role.read:

139 result = True

             elif isinstance(object, Photo):

                 acl = self.get_acl(user, object)

                 if acl is not None and acl > Role.read:

143 result = True

144 else:

                     albums = self.get_albums(object)

                     acl = self.get_acl(user, albums[0])

                     if acl is not None and acl > Role.read:

148 result = True

149 else:

                         portfolio = self.get_portfolios(object)

                         acl = self.get_acl(user, portfolio[0])

                         if acl is not None and acl > Role.read:

153 result = True

154 else:

                 raise DBQueryException("%s: not supported" % object.__tablename__)

156 return result

157 else:

             raise DBQueryException("{}: not supported".format(object))

159

     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):

164 return Role.admin

                 acl = AccessRight.query.join(AccessRight.album).filter(Album.id == object.id)

166 for a in acl:

                     if AccessRight.query.get(a.id).user_id == user.id:

168 return a.right

169 # 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):

173 return Role.admin

                 acl = AccessRight.query.join(AccessRight.photo).filter(Photo.id == object.id)

175 for a in acl:

                     if AccessRight.query.get(a.id).user_id == user.id:

177 return a.right

178 # 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):

182 return Role.admin

                 acl = AccessRight.query.join(AccessRight.portfolio).filter(Portfolio.id == object.id)

184 for a in acl:

                     if AccessRight.query.get(a.id).user_id == user.id:

186 return a.right

187 else:

                 raise DBQueryException("%s: not supported" % object.__tablename__)

189 return None

190 else:

             raise DBQueryException("{}: not supported".format(object))

192

     def get_albums_for_user(self, user, owner = False, hidden = False):

         if isinstance(user, User):

195 albums = []

196 # 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()

202 for object in objects:

                 if not owner and object.Portfolio.owner == user:

204 continue

                 if not hidden and not object.Album.visible:

206 continue

                 albums.append(object.Album)

208 # Find all albums with access through inherited access rights from portfolio

209 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()

214 for object in objects:

                 if not owner and object.Portfolio.owner == user:

216 continue

                 if not hidden and not object.Album.visible:

218 continue

                 temp.append(object.Album)

220 # Merge albums with temp removing duplicates

             albums = albums + list(set(temp) - set(albums))

222 return albums

223 else:

             raise DBQueryException("%s: not supported" % object.__tablename__)