Package coprs :: Package logic :: Module users_logic
[hide private]
[frames] | no frames]

Source Code for Module coprs.logic.users_logic

  1  import json 
  2  import simplejson 
  3  from coprs import exceptions 
  4  from flask import url_for 
  5   
  6  from coprs import app, db 
  7  from coprs.models import User, Group 
  8  from coprs.helpers import copr_url 
  9  from sqlalchemy import update 
10 11 12 -class UsersLogic(object):
13 14 @classmethod
15 - def get(cls, username):
16 return User.query.filter(User.username == username)
17 18 @classmethod
19 - def get_by_api_login(cls, login):
20 return User.query.filter(User.api_login == login)
21 22 @classmethod
23 - def raise_if_cant_update_copr(cls, user, copr, message):
24 """ 25 Raise InsufficientRightsException if given user cant update 26 given copr. Return None otherwise. 27 """ 28 29 # TODO: this is a bit inconsistent - shouldn't the user method be 30 # called can_update? 31 if not user.can_edit(copr): 32 raise exceptions.InsufficientRightsException(message)
33 34 @classmethod
35 - def raise_if_cant_build_in_copr(cls, user, copr, message):
36 """ 37 Raises InsufficientRightsException if given user cant build in 38 given copr. Return None otherwise. 39 """ 40 41 if not user.can_build_in(copr): 42 raise exceptions.InsufficientRightsException(message)
43 44 @classmethod
45 - def raise_if_not_in_group(cls, user, group):
46 if group.fas_name not in user.user_teams: 47 raise exceptions.InsufficientRightsException( 48 "User '{}' doesn't have access to group {}({})" 49 .format(user.username, group.name, group.fas_name))
50 51 @classmethod
52 - def get_group_by_alias(cls, name):
53 return Group.query.filter(Group.name == name)
54 55 @classmethod
56 - def group_alias_exists(cls, name):
57 query = cls.get_group_by_alias(name) 58 return query.count() != 0
59 60 @classmethod
61 - def get_group_by_fas_name(cls, fas_name):
62 return Group.query.filter(Group.fas_name == fas_name)
63 64 @classmethod
65 - def get_groups_by_fas_names_list(cls, fas_name_list):
66 return Group.query.filter(Group.fas_name.in_(fas_name_list))
67 68 @classmethod
69 - def get_groups_by_names_list(cls, name_list):
70 return Group.query.filter(Group.name.in_(name_list))
71 72 @classmethod
73 - def create_group_by_fas_name(cls, fas_name, alias=None):
74 if alias is None: 75 alias = fas_name 76 77 group = Group( 78 fas_name=fas_name, 79 name=alias, 80 ) 81 db.session.add(group) 82 return group
83 84 @classmethod
85 - def get_group_by_fas_name_or_create(cls, fas_name, alias=None):
86 mb_group = cls.get_group_by_fas_name(fas_name).first() 87 if mb_group is not None: 88 return mb_group 89 90 group = cls.create_group_by_fas_name(fas_name, alias) 91 db.session.flush() 92 return group
93 94 @classmethod
95 - def filter_blacklisted_teams(cls, teams):
96 """ removes blacklisted groups from teams list 97 :type teams: list of str 98 :return: filtered teams 99 :rtype: list of str 100 """ 101 blacklist = set(app.config.get("BLACKLISTED_GROUPS", [])) 102 return filter(lambda t: t not in blacklist, teams)
103 104 @classmethod
105 - def is_blacklisted_group(cls, fas_group):
106 if "BLACKLISTED_GROUPS" in app.config: 107 return fas_group in app.config["BLACKLISTED_GROUPS"] 108 else: 109 return False
110 111 @classmethod
112 - def delete_user_data(cls, fas_name):
113 query = update(User).where(User.username==fas_name).\ 114 values( 115 timezone=None, 116 proven=False, 117 admin=False, 118 proxy=False, 119 api_login='', 120 api_token='', 121 api_token_expiration='1970-01-01', 122 openid_groups=None 123 ) 124 db.engine.connect().execute(query)
125
126 127 -class UserDataDumper(object):
128 - def __init__(self, user):
129 self.user = user
130
131 - def dumps(self, pretty=False):
132 if pretty: 133 return simplejson.dumps(self.data, indent=2) 134 return json.dumps(self.data)
135 136 @property
137 - def data(self):
138 data = self.user_information 139 data["groups"] = self.groups 140 data["projects"] = self.projects 141 data["builds"] = self.builds 142 return data
143 144 @property
145 - def user_information(self):
146 return { 147 "username": self.user.name, 148 "email": self.user.mail, 149 "timezone": self.user.timezone, 150 "api_login": self.user.api_login, 151 "api_token": self.user.api_token, 152 "api_token_expiration": self.user.api_token_expiration.strftime("%b %d %Y %H:%M:%S"), 153 "gravatar": self.user.gravatar_url, 154 }
155 156 @property
157 - def groups(self):
158 return [{"name": g.name, 159 "url": url_for("groups_ns.list_projects_by_group", group_name=g.name, _external=True)} 160 for g in self.user.user_groups]
161 162 @property
163 - def projects(self):
164 # @FIXME We get into circular import when this import is on module-level 165 from coprs.logic.coprs_logic import CoprsLogic 166 return [{"full_name": p.full_name, 167 "url": copr_url("coprs_ns.copr_detail", p, _external=True)} 168 for p in CoprsLogic.filter_by_user_name(CoprsLogic.get_multiple(), self.user.name)]
169 170 @property
171 - def builds(self):
172 return [{"id": b.id, 173 "project": b.copr.full_name, 174 "url": copr_url("coprs_ns.copr_build", b.copr, build_id=b.id, _external=True)} 175 for b in self.user.builds]
176