sqlalchemy.py 文件源码

def get_object(self, req, resp, path_params, for_update=False, db_session=None):
        """
        :param req: Falcon request
        :type req: falcon.request.Request

        :param resp: Falcon response
        :type resp: falcon.response.Response

        :param path_params: path params extracted from URL path
        :type path_params: dict

        :param for_update: if the object is going to be updated or deleted
        :type for_update: bool

        :param db_session: SQLAlchemy session
        :type db_session: sqlalchemy.orm.session.Session
        """
        query = db_session.query(self.objects_class)
        if for_update:
            query = query.with_for_update()

        for key, value in path_params.items():
            attr = getattr(self.objects_class, key, None)
            query = query.filter(attr == value)

        conditions = dict(req.params)
        if self.PARAM_RELATIONS in conditions:
            conditions.pop(self.PARAM_RELATIONS)
        query = self.filter_by(query, conditions)

        try:
            obj = query.one()
        except NoResultFound:
            raise HTTPNotFound()
        except MultipleResultsFound:
            raise HTTPBadRequest('Multiple results', 'Query params match multiple records')
        return obj