Added BaseDomainClass with a get method.

The BaseDomainClass.get can accept any number of keyword arguments and
will add a filter to the query for each kwarg. This will allow the
caller to scope the query as needed with kwargs.
This commit is contained in:
dandds 2019-04-16 10:05:06 -04:00
parent 5ea70a486a
commit eaeeed0b05
4 changed files with 46 additions and 26 deletions

View File

@ -0,0 +1,25 @@
from sqlalchemy.orm.exc import NoResultFound
from atst.database import db
from atst.domain.exceptions import NotFoundError
class BaseDomainClass(object):
model = None
resource_name = None
@classmethod
def get(cls, resource_id, **kwargs):
base_query = db.session.query(cls.model).filter(cls.model.id == resource_id)
if getattr(cls.model, "deleted", False):
base_query = base_query.filter(cls.model.deleted == False)
for col, val in kwargs.items():
base_query = base_query.filter(getattr(cls.model, col) == val)
try:
resource = base_query.one()
return resource
except NoResultFound:
raise NotFoundError(cls.resource_name)

View File

@ -1,6 +1,7 @@
from sqlalchemy.orm.exc import NoResultFound from sqlalchemy.orm.exc import NoResultFound
from atst.database import db from atst.database import db
from . import BaseDomainClass
from atst.domain.environments import Environments from atst.domain.environments import Environments
from atst.domain.exceptions import NotFoundError from atst.domain.exceptions import NotFoundError
from atst.models.application import Application from atst.models.application import Application
@ -8,7 +9,10 @@ from atst.models.environment import Environment
from atst.models.environment_role import EnvironmentRole from atst.models.environment_role import EnvironmentRole
class Applications(object): class Applications(BaseDomainClass):
model = Application
resource_name = "application"
@classmethod @classmethod
def create(cls, portfolio, name, description, environment_names): def create(cls, portfolio, name, description, environment_names):
application = Application( application = Application(
@ -21,19 +25,6 @@ class Applications(object):
db.session.commit() db.session.commit()
return application return application
@classmethod
def get(cls, application_id):
try:
application = (
db.session.query(Application)
.filter_by(id=application_id, deleted=False)
.one()
)
except NoResultFound:
raise NotFoundError("application")
return application
@classmethod @classmethod
def for_user(self, user, portfolio): def for_user(self, user, portfolio):
return ( return (

View File

@ -1,19 +1,21 @@
from sqlalchemy.orm.exc import NoResultFound
from flask import current_app as app from flask import current_app as app
from atst.database import db from atst.database import db
from atst.models.task_order import TaskOrder from atst.models.task_order import TaskOrder
from atst.models.dd_254 import DD254 from atst.models.dd_254 import DD254
from . import BaseDomainClass
from atst.domain.portfolios import Portfolios from atst.domain.portfolios import Portfolios
from atst.domain.permission_sets import PermissionSets from atst.domain.permission_sets import PermissionSets
from .exceptions import NotFoundError
class TaskOrderError(Exception): class TaskOrderError(Exception):
pass pass
class TaskOrders(object): class TaskOrders(BaseDomainClass):
model = TaskOrder
resource_name = "task_order"
SECTIONS = { SECTIONS = {
"app_info": [ "app_info": [
"portfolio_name", "portfolio_name",
@ -51,15 +53,6 @@ class TaskOrders(object):
UNCLASSIFIED_FUNDING = ["performance_length", "csp_estimate", "clin_01", "clin_03"] UNCLASSIFIED_FUNDING = ["performance_length", "csp_estimate", "clin_01", "clin_03"]
@classmethod
def get(cls, task_order_id):
try:
task_order = db.session.query(TaskOrder).filter_by(id=task_order_id).one()
return task_order
except NoResultFound:
raise NotFoundError("task_order")
@classmethod @classmethod
def create(cls, creator, portfolio): def create(cls, creator, portfolio):
task_order = TaskOrder(portfolio=portfolio, creator=creator) task_order = TaskOrder(portfolio=portfolio, creator=creator)

View File

@ -1,4 +1,5 @@
import pytest import pytest
from uuid import uuid4
from atst.domain.applications import Applications from atst.domain.applications import Applications
from atst.domain.exceptions import NotFoundError from atst.domain.exceptions import NotFoundError
@ -70,6 +71,16 @@ def test_get_excludes_deleted():
Applications.get(app.id) Applications.get(app.id)
def test_get_application():
app = ApplicationFactory.create()
assert Applications.get(app.id) == app
assert Applications.get(app.id, portfolio_id=app.portfolio_id) == app
with pytest.raises(NotFoundError):
# make the uuid a string like you'd get from a route
rando_id = str(uuid4())
Applications.get(app.id, portfolio_id=rando_id)
def test_delete_application(session): def test_delete_application(session):
app = ApplicationFactory.create() app = ApplicationFactory.create()
app_role = ApplicationRoleFactory.create(user=UserFactory.create(), application=app) app_role = ApplicationRoleFactory.create(user=UserFactory.create(), application=app)