Trees | Indices | Help |
---|
|
1 import time 2 3 from coprs import db 4 from coprs import exceptions 5 from coprs import helpers 6 from coprs import models 7 from coprs import signals 8 from coprs.logic import users_logic11 12 """ 13 Used for manipulating Coprs. 14 15 All methods accept user object as a first argument, 16 as this may be needed in future. 17 """ 18 19 @classmethod25021 """ Return all coprs without those which are deleted. """ 22 query = (db.session.query(models.Copr) 23 .join(models.Copr.owner) 24 .options(db.contains_eager(models.Copr.owner)) 25 .filter(models.Copr.deleted == False)) 26 return query27 28 @classmethod30 with_builds = kwargs.get("with_builds", False) 31 with_mock_chroots = kwargs.get("with_mock_chroots", False) 32 33 query = (cls.get_all() 34 .filter(models.Copr.name == coprname) 35 .filter(models.User.openid_name == 36 models.User.openidize_name(username)) 37 ) 38 39 if with_builds: 40 query = (query.outerjoin(models.Copr.builds) 41 .options(db.contains_eager(models.Copr.builds)) 42 .order_by(models.Build.submitted_on.desc())) 43 44 if with_mock_chroots: 45 query = (query.outerjoin(*models.Copr.mock_chroots.attr) 46 .options(db.contains_eager(*models.Copr.mock_chroots.attr)) 47 .order_by(models.MockChroot.os_release.asc()) 48 .order_by(models.MockChroot.os_version.asc()) 49 .order_by(models.MockChroot.arch.asc())) 50 51 return query52 53 @classmethod55 user_relation = kwargs.get("user_relation", None) 56 username = kwargs.get("username", None) 57 with_mock_chroots = kwargs.get("with_mock_chroots", None) 58 with_builds = kwargs.get("with_builds", None) 59 incl_deleted = kwargs.get("incl_deleted", None) 60 ids = kwargs.get("ids", None) 61 62 query = (db.session.query(models.Copr) 63 .join(models.Copr.owner) 64 .options(db.contains_eager(models.Copr.owner)) 65 .order_by(models.Copr.id.desc())) 66 67 if not incl_deleted: 68 query = query.filter(models.Copr.deleted == False) 69 70 if isinstance(ids, list): # can be an empty list 71 query = query.filter(models.Copr.id.in_(ids)) 72 73 if user_relation == "owned": 74 query = query.filter( 75 models.User.openid_name == models.User.openidize_name(username)) 76 elif user_relation == "allowed": 77 aliased_user = db.aliased(models.User) 78 79 query = (query.join(models.CoprPermission, 80 models.Copr.copr_permissions) 81 .filter(models.CoprPermission.copr_builder == 82 helpers.PermissionEnum('approved')) 83 .join(aliased_user, models.CoprPermission.user) 84 .filter(aliased_user.openid_name == 85 models.User.openidize_name(username))) 86 87 if with_mock_chroots: 88 query = (query.outerjoin(*models.Copr.mock_chroots.attr) 89 .options(db.contains_eager(*models.Copr.mock_chroots.attr)) 90 .order_by(models.MockChroot.os_release.asc()) 91 .order_by(models.MockChroot.os_version.asc()) 92 .order_by(models.MockChroot.arch.asc())) 93 94 if with_builds: 95 query = (query.outerjoin(models.Copr.builds) 96 .options(db.contains_eager(models.Copr.builds)) 97 .order_by(models.Build.submitted_on.desc())) 98 99 return query100 101 @classmethod 104 105 @classmethod107 if user.admin: 108 db.session.add(copr) 109 pass 110 else: 111 raise exceptions.InsufficientRightsException( 112 "User is not a system admin")113 114 @classmethod116 query = (models.Copr.query.join(models.User) 117 .filter(models.Copr.deleted == False) 118 .whooshee_search(search_string)) 119 return query120 121 @classmethod122 - def add(cls, user, name, repos, selected_chroots, description, 123 instructions, check_for_duplicates=False):124 copr = models.Copr(name=name, 125 repos=repos, 126 owner=user, 127 description=description, 128 instructions=instructions, 129 created_on=int(time.time())) 130 131 # form validation checks for duplicates 132 CoprsLogic.new(user, copr, 133 check_for_duplicates=check_for_duplicates) 134 CoprChrootsLogic.new_from_names(user, copr, 135 selected_chroots) 136 return copr137 138 @classmethod140 if check_for_duplicates and cls.exists_for_user(user, copr.name).all(): 141 raise exceptions.DuplicateException( 142 "Copr: '{0}' already exists".format(copr.name)) 143 signals.copr_created.send(cls, copr=copr) 144 db.session.add(copr)145 146 @classmethod148 cls.raise_if_unfinished_blocking_action( 149 user, copr, "Can't change this project name," 150 " another operation is in progress: {action}") 151 152 users_logic.UsersLogic.raise_if_cant_update_copr( 153 user, copr, "Only owners and admins may update their projects.") 154 155 existing = cls.exists_for_user(copr.owner, copr.name).first() 156 if existing: 157 if check_for_duplicates and existing.id != copr.id: 158 raise exceptions.DuplicateException( 159 "Project: '{0}' already exists".format(copr.name)) 160 161 else: # we're renaming 162 # if we fire a models.Copr.query, it will use the modified copr in session 163 # -> workaround this by just getting the name 164 old_copr_name = (db.session.query(models.Copr.name) 165 .filter(models.Copr.id == copr.id) 166 .filter(models.Copr.deleted == False) 167 .first()[0]) 168 169 action = models.Action(action_type=helpers.ActionTypeEnum("rename"), 170 object_type="copr", 171 object_id=copr.id, 172 old_value="{0}/{1}".format(copr.owner.name, 173 old_copr_name), 174 new_value="{0}/{1}".format(copr.owner.name, 175 copr.name), 176 created_on=int(time.time())) 177 db.session.add(action) 178 db.session.add(copr)179 180 @classmethod182 cls.raise_if_cant_delete(user, copr) 183 # TODO: do we want to dump the information somewhere, so that we can 184 # search it in future? 185 cls.raise_if_unfinished_blocking_action( 186 user, copr, "Can't delete this project," 187 " another operation is in progress: {action}") 188 189 action = models.Action(action_type=helpers.ActionTypeEnum("delete"), 190 object_type="copr", 191 object_id=copr.id, 192 old_value="{0}/{1}".format(copr.owner.name, 193 copr.name), 194 new_value="", 195 created_on=int(time.time())) 196 copr.deleted = True 197 198 db.session.add(action) 199 200 return copr201 202 @classmethod204 existing = (models.Copr.query 205 .filter(models.Copr.name == coprname) 206 .filter(models.Copr.owner_id == user.id)) 207 208 if not incl_deleted: 209 existing = existing.filter(models.Copr.deleted == False) 210 211 return existing212 213 @classmethod215 blocking_actions = [helpers.ActionTypeEnum("rename"), 216 helpers.ActionTypeEnum("delete")] 217 218 actions = (models.Action.query 219 .filter(models.Action.object_type == "copr") 220 .filter(models.Action.object_id == copr.id) 221 .filter(models.Action.result == 222 helpers.BackendResultEnum("waiting")) 223 .filter(models.Action.action_type.in_(blocking_actions))) 224 225 return actions226 227 @classmethod229 """ 230 Raise ActionInProgressException if given copr has an unfinished 231 action. Return None otherwise. 232 """ 233 234 unfinished_actions = cls.unfinished_blocking_actions_for( 235 user, copr).all() 236 if unfinished_actions: 237 raise exceptions.ActionInProgressException( 238 message, unfinished_actions[0])239 240 @classmethod242 """ 243 Raise InsufficientRightsException if given copr cant be deleted 244 by given user. Return None otherwise. 245 """ 246 247 if not user.admin and user != copr.owner: 248 raise exceptions.InsufficientRightsException( 249 "Only owners may delete their projects.")253 254 @classmethod312256 query = (models.CoprPermission.query 257 .filter(models.CoprPermission.copr == copr) 258 .filter(models.CoprPermission.user == searched_user)) 259 260 return query261 262 @classmethod264 query = models.CoprPermission.query.filter( 265 models.CoprPermission.copr == copr) 266 267 return query268 269 @classmethod 272 273 @classmethod276 277 users_logic.UsersLogic.raise_if_cant_update_copr( 278 user, copr, "Only owners and admins may update" 279 " their projects permissions.") 280 281 (models.CoprPermission.query 282 .filter(models.CoprPermission.copr_id == copr.id) 283 .filter(models.CoprPermission.user_id == copr_permission.user_id) 284 .update({"copr_builder": new_builder, 285 "copr_admin": new_admin}))286 287 @classmethod289 if copr_permission: 290 # preserve approved permissions if set 291 if (not new_builder or copr_permission.copr_builder != 292 helpers.PermissionEnum("approved")): 293 294 copr_permission.copr_builder = new_builder 295 296 if (not new_admin or copr_permission.copr_admin != 297 helpers.PermissionEnum("approved")): 298 299 copr_permission.copr_admin = new_admin 300 else: 301 perm = models.CoprPermission( 302 user=user, 303 copr=copr, 304 copr_builder=new_builder, 305 copr_admin=new_admin) 306 307 cls.new(user, perm)308 309 @classmethod315 316 @classmethod363318 db_chroots = models.MockChroot.query.all() 319 mock_chroots = [] 320 for ch in db_chroots: 321 if ch.name in names: 322 mock_chroots.append(ch) 323 324 return mock_chroots325 326 @classmethod 329 330 @classmethod332 for mock_chroot in cls.mock_chroots_from_names(user, names): 333 db.session.add( 334 models.CoprChroot(copr=copr, mock_chroot=mock_chroot))335 336 @classmethod338 copr_chroot = copr.check_copr_chroot(chroot) 339 if copr_chroot: 340 copr_chroot.buildroot_pkgs = buildroot_pkgs 341 db.session.add(copr_chroot)342 343 @classmethod345 current_chroots = copr.mock_chroots 346 new_chroots = cls.mock_chroots_from_names(user, names) 347 # add non-existing 348 for mock_chroot in new_chroots: 349 if mock_chroot not in current_chroots: 350 db.session.add( 351 models.CoprChroot(copr=copr, mock_chroot=mock_chroot)) 352 353 # delete no more present 354 to_remove = [] 355 for mock_chroot in current_chroots: 356 if mock_chroot not in new_chroots: 357 # can't delete here, it would change current_chroots and break 358 # iteration 359 to_remove.append(mock_chroot) 360 361 for mc in to_remove: 362 copr.mock_chroots.remove(mc)366 367 @classmethod464369 if noarch and not arch: 370 return (models.MockChroot.query 371 .filter(models.MockChroot.os_release == os_release, 372 models.MockChroot.os_version == os_version)) 373 374 return (models.MockChroot.query 375 .filter(models.MockChroot.os_release == os_release, 376 models.MockChroot.os_version == os_version, 377 models.MockChroot.arch == arch))378 379 @classmethod381 """ 382 chroot_name should be os-version-architecture, e.g. fedora-rawhide-x86_64 383 the architecture could be optional with noarch=True 384 385 Return MockChroot object for textual representation of chroot 386 """ 387 388 name_tuple = cls.tuple_from_name(None, chroot_name, noarch=noarch) 389 return cls.get(None, name_tuple[0], name_tuple[1], name_tuple[2], 390 active_only=active_only, noarch=noarch)391 392 @classmethod394 query = models.MockChroot.query 395 if active_only: 396 query = query.filter(models.MockChroot.is_active == True) 397 return query398 399 @classmethod401 name_tuple = cls.tuple_from_name(user, name) 402 if cls.get(user, *name_tuple).first(): 403 raise exceptions.DuplicateException( 404 "Mock chroot with this name already exists.") 405 new_chroot = models.MockChroot(os_release=name_tuple[0], 406 os_version=name_tuple[1], 407 arch=name_tuple[2]) 408 cls.new(user, new_chroot) 409 return new_chroot410 411 @classmethod 414 415 @classmethod417 name_tuple = cls.tuple_from_name(user, name) 418 mock_chroot = cls.get(user, *name_tuple).first() 419 if not mock_chroot: 420 raise exceptions.NotFoundException( 421 "Mock chroot with this name doesn't exist.") 422 423 mock_chroot.is_active = is_active 424 cls.update(user, mock_chroot) 425 return mock_chroot426 427 @classmethod 430 431 @classmethod433 name_tuple = cls.tuple_from_name(user, name) 434 mock_chroot = cls.get(user, *name_tuple).first() 435 if not mock_chroot: 436 raise exceptions.NotFoundException( 437 "Mock chroot with this name doesn't exist.") 438 439 cls.delete(user, mock_chroot)440 441 @classmethod 444 445 @classmethod447 """ 448 input should be os-version-architecture, e.g. fedora-rawhide-x86_64 449 450 the architecture could be optional with noarch=True 451 452 returns ("os", "vetsion", "arch") or ("os", "version", None) 453 """ 454 splitted_name = name.split("-") 455 if (not noarch and len(splitted_name) is not 3)\ 456 or (noarch and len(splitted_name) not in (2,3)): 457 raise exceptions.MalformedArgumentException( 458 "Chroot name is not valid") 459 460 if noarch: 461 splitted_name.append(None) 462 463 return tuple(splitted_name)
Trees | Indices | Help |
---|
Generated by Epydoc 3.0.1 on Wed Aug 27 09:50:36 2014 | http://epydoc.sourceforge.net |