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

Source Code for Module coprs.logic.users_logic

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