Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_without_collection_path_has_one_service(self, mocked_service):
@resource(path='/nocollection/{id}', name='nocollection')
class NoCollection(object):
def __init__(self, request, context=None):
pass
self.assertEqual(mocked_service.call_count, 1)
from cornice.resource import resource, view
from ode.models import Source
from ode.resources.base import ResourceMixin, set_content_type
from ode.resources.base import COLLECTION_JSON_MIMETYPE
from ode.validation.schema import SourceCollectionSchema
from ode.validation.validators import has_provider_id
from ode.validation.validators import validate_querystring
@resource(collection_path='/v1/sources', path='/v1/sources/{id}',
filters=set_content_type)
class SourceResource(ResourceMixin):
model = Source
@view(validators=[has_provider_id], schema=SourceCollectionSchema,
content_type=COLLECTION_JSON_MIMETYPE)
def collection_post(self):
return ResourceMixin.collection_post(self)
@view(validators=[has_provider_id], schema=SourceCollectionSchema,
content_type=COLLECTION_JSON_MIMETYPE)
def put(self):
return ResourceMixin.put(self)
@view(validators=[has_provider_id, validate_querystring])
import json
import os
from cornice.resource import resource, view
from webgnome.schema import ModelSchema
from webgnome.views.services.base import BaseResource
from webgnome import util
@resource(path='/model/{model_id}/location_file/{location}',
renderer='gnome_json',
description='Add a location file to the current model')
class LocationFile(BaseResource):
@view(validators=util.valid_location_file)
def post(self):
data = self.request.validated
model = data['model']
raw_data = open(data['location_file']).read()
location_data = json.loads(raw_data)
validated = ModelSchema().bind().deserialize(location_data)
_map = validated.get('map', None)
if _map and _map['filename']:
_map['filename'] = os.path.join(data['location_dir'],
_map['filename'])
from cornice.resource import resource, view
from webgnome import util
from webgnome import schema
from webgnome.model_manager import WebWind
from webgnome.views.services.base import BaseResource
@resource(collection_path='/model/{model_id}/environment/wind',
path='/model/{model_id}/environment/wind/{id}',
renderer='gnome_json', description='A wind mover.')
class Wind(BaseResource):
@view(validators=util.valid_model_id, schema=schema.WindSchema)
def collection_post(self):
"""
Create a :class:`model_manager.WebWind` from a JSON representation.
"""
data = self.request.validated
model = data.pop('model')
wind = WebWind(**data)
model.environment.add(wind)
wind_data = wind.to_dict()
return schema.WindSchema().bind().serialize(wind_data)
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
"""Defines service endpoints pertaining to Composes."""
from cornice.resource import resource, view
from pyramid import httpexceptions
from pyramid.security import Allow, Everyone
from sqlalchemy.orm import exc
from bodhi.server import models, security
from bodhi.server.services import errors
@resource(collection_path='/composes/', path='/composes/{release_name}/{request}',
description='Compose service')
class Composes(object):
"""
Defines resources for interacting with Compose objects.
Operations acting on the collection are served at ``/composes/`` and operations acting on a
single compose are served at ``/composes//``.
"""
def __init__(self, request, context=None):
"""
Initialize the Compose resource.
Args:
request (pyramid.request.Request): The current web request.
context (None): Unused.
from openprocurement.api.utils import (
apply_patch,
save_tender,
error_handler,
update_journal_handler_params,
)
from openprocurement.api.validation import (
validate_contract_data,
validate_patch_contract_data,
)
LOGGER = getLogger(__name__)
@resource(name='Tender Award Contracts',
collection_path='/tenders/{tender_id}/awards/{award_id}/contracts',
path='/tenders/{tender_id}/awards/{award_id}/contracts/{contract_id}',
description="Tender award contracts",
error_handler=error_handler)
class TenderAwardContractResource(object):
def __init__(self, request):
self.request = request
self.db = request.registry.db
@view(content_type="application/json", permission='create_award_contract', validators=(validate_contract_data,), renderer='json')
def collection_post(self):
"""Post a contract for award
"""
tender = self.request.validated['tender']
if tender.status not in ['active.awarded', 'complete']:
wind = data.pop('wind')
mover.from_dict(data)
mover.wind = wind
mover_data = mover.to_dict(do='create')
return schema.WindMoverSchema().bind().serialize(mover_data)
@view(validators=util.valid_mover_id)
def delete(self):
"""
Delete a :class:`model_manager.WebWindMover`.
"""
self.request.validated['model'].movers.remove(self.id)
@resource(collection_path='/model/{model_id}/mover/random',
path='/model/{model_id}/mover/random/{id}',
renderer='gnome_json', description='A random mover.')
class RandomMover(BaseResource):
@view(validators=util.valid_model_id,
schema=schema.RandomMoverSchema)
def collection_post(self):
"""
Create a :class:`model_manager.WebRandomMover` from a JSON
representation.
"""
data = self.request.validated
model = data.pop('model')
mover = WebRandomMover(**data)
model.movers.add(mover)
mover_data = mover.to_dict(do='create')
def request_params(request):
try:
params = NestedMultiDict(request.GET, request.POST)
except UnicodeDecodeError:
request.errors.add('body', 'data', 'could not decode params')
request.errors.status = 422
raise error_handler(request, False)
except Exception as e:
request.errors.add('body', str(e.__class__.__name__), str(e))
request.errors.status = 422
raise error_handler(request, False)
return params
opresource = partial(resource, error_handler=error_handler, factory=factory)
def forbidden(request):
request.errors.add('url', 'permission', 'Forbidden')
request.errors.status = 403
return error_handler(request)
def update_logging_context(request, params):
if not request.__dict__.get('logging_context'):
request.logging_context = {}
for x, j in params.items():
request.logging_context[x.upper()] = j
from cornice.resource import resource, view
from cms.api import validators, utils
from gitmodel.exceptions import DoesNotExist
@resource(
collection_path='/api/categories.json',
path='/api/categories/{uuid}.json'
)
class CategoryApi(utils.ApiBase):
def collection_get(self):
models = self.get_repo_models()
return [c.to_dict() for c in models.GitCategoryModel().all()]
@view(renderer='json')
def get(self):
models = self.get_repo_models()
uuid = self.request.matchdict['uuid']
try:
category = models.GitCategoryModel().get(uuid)
return category.to_dict()
self.request.errors.status = 403
return
src = self.tender.serialize("plain")
res = self.bid.serialize("view")
self.tender.bids.remove(self.bid)
patch = make_patch(self.tender.serialize("plain"), src).patch
if patch:
self.tender.revisions.append(revision({'changes': patch}))
try:
self.tender.store(self.db)
except Exception, e:
return self.request.errors.add('body', 'data', str(e))
return {'data': res}
@resource(name='Tender Bid Documents',
collection_path='/tenders/{tender_id}/bidders/{bid_id}/documents',
path='/tenders/{tender_id}/bidders/{bid_id}/documents/{id}',
description="Tender bidder documents")
class TenderBidderDocumentResource(object):
def __init__(self, request):
self.request = request
self.db = request.registry.db
self.tender_id = request.matchdict['tender_id']
self.bid_id = request.matchdict['bid_id']
self.document_id = self.request.matchdict.get('id')
self.tender = TenderDocument.load(self.db, self.tender_id)
if self.tender:
bids = [i for i in self.tender.bids if i.id == self.bid_id]
self.bid = bids and bids[0]
if self.bid and self.document_id: