Create utility function for the pattern of committing to the database or raising AlreadyExistsError

This commit is contained in:
leigh-mil 2019-12-16 10:39:47 -05:00
parent afad5362a1
commit b927ef1b0e
3 changed files with 20 additions and 33 deletions

View File

@ -1,7 +1,4 @@
from sqlalchemy.exc import IntegrityError
from . import BaseDomainClass
from .exceptions import AlreadyExistsError
from flask import g
from atst.database import db
from atst.domain.application_roles import ApplicationRoles
@ -14,7 +11,7 @@ from atst.models import (
ApplicationRoleStatus,
EnvironmentRole,
)
from atst.utils import first_or_none
from atst.utils import first_or_none, update_or_raise_already_exists_error
class Applications(BaseDomainClass):
@ -31,12 +28,7 @@ class Applications(BaseDomainClass):
if environment_names:
Environments.create_many(user, application, environment_names)
try:
db.session.commit()
except IntegrityError:
db.session.rollback()
raise AlreadyExistsError("application")
update_or_raise_already_exists_error(message="application")
return application
@classmethod
@ -61,14 +53,9 @@ class Applications(BaseDomainClass):
Environments.create_many(
g.current_user, application, new_data["environment_names"]
)
db.session.add(application)
try:
db.session.commit()
except IntegrityError:
db.session.rollback()
raise AlreadyExistsError("application")
update_or_raise_already_exists_error(message="application")
return application
@classmethod

View File

@ -1,11 +1,10 @@
import datetime
from sqlalchemy.exc import IntegrityError
from atst.database import db
from atst.models.clin import CLIN
from atst.models.task_order import TaskOrder, SORT_ORDERING
from . import BaseDomainClass
from .exceptions import AlreadyExistsError
from atst.utils import update_or_raise_already_exists_error
class TaskOrders(BaseDomainClass):
@ -16,15 +15,8 @@ class TaskOrders(BaseDomainClass):
def create(cls, portfolio_id, number, clins, pdf):
task_order = TaskOrder(portfolio_id=portfolio_id, number=number, pdf=pdf)
db.session.add(task_order)
try:
db.session.commit()
except IntegrityError:
db.session.rollback()
raise AlreadyExistsError("task_order")
update_or_raise_already_exists_error(message="task_order")
TaskOrders.create_clins(task_order.id, clins)
return task_order
@classmethod
@ -42,12 +34,7 @@ class TaskOrders(BaseDomainClass):
task_order.number = number
db.session.add(task_order)
try:
db.session.commit()
except IntegrityError:
db.session.rollback()
raise AlreadyExistsError("task_order")
update_or_raise_already_exists_error(message="task_order")
return task_order
@classmethod

View File

@ -1,5 +1,10 @@
import re
from sqlalchemy.exc import IntegrityError
from atst.database import db
from atst.domain.exceptions import AlreadyExistsError
def first_or_none(predicate, lst):
return next((x for x in lst if predicate(x)), None)
@ -23,3 +28,11 @@ def camel_to_snake(camel_cased):
def pick(keys, dct):
_keys = set(keys)
return {k: v for (k, v) in dct.items() if k in _keys}
def update_or_raise_already_exists_error(message):
try:
db.session.commit()
except IntegrityError:
db.session.rollback()
raise AlreadyExistsError(message)