How to use _fresh_queryset method in autotest

Best Python code snippet using autotest_python

resource_lib.py

Source:resource_lib.py Github

copy

Full Screen

...272class Collection(Resource):273 _DEFAULT_ITEMS_PER_PAGE = 50274 _permitted_methods=('GET', 'POST')275 # subclasses must override these276 queryset = None # or override _fresh_queryset() directly277 entry_class = None278 def __init__(self, request):279 super(Collection, self).__init__(request)280 assert self.entry_class is not None281 if isinstance(self.entry_class, basestring):282 type(self).entry_class = _resolve_class_path(self.entry_class)283 self._query_processor = query_lib.QueryProcessor()284 self.entry_class.add_query_selectors(self._query_processor)285 def _query_parameters_accepted(self):286 params = [('start_index', 'Index of first member to include'),287 ('items_per_page', 'Number of members to include'),288 ('full_representations',289 'True to include full representations of members')]290 for selector in self._query_processor.selectors():291 params.append((selector.name, selector.doc))292 return params293 def _fresh_queryset(self):294 assert self.queryset is not None295 # always copy the queryset before using it to avoid caching296 return self.queryset.all()297 def _entry_from_instance(self, instance):298 return self.entry_class(self._request, instance)299 def _representation(self, entry_instances):300 entries = [self._entry_from_instance(instance)301 for instance in entry_instances]302 want_full_representation = self._read_bool_parameter(303 'full_representations')304 if want_full_representation:305 self.entry_class.prepare_for_full_representation(entries)306 members = []307 for entry in entries:308 if want_full_representation:309 rep = entry.full_representation()310 else:311 rep = entry.short_representation()312 members.append(rep)313 rep = self.link()314 rep.update({'members': members})315 return rep316 def _read_bool_parameter(self, name):317 if name not in self._query_params:318 return False319 return (self._query_params[name].lower() == 'true')320 def _read_int_parameter(self, name, default):321 if name not in self._query_params:322 return default323 input_value = self._query_params[name]324 try:325 return int(input_value)326 except ValueError:327 raise exceptions.BadRequest('Invalid non-numeric value for %s: %r'328 % (name, input_value))329 def _apply_form_query(self, queryset):330 """Apply any query selectors passed as form variables."""331 for parameter, values in self._query_params.lists():332 if ':' in parameter:333 parameter, comparison_type = parameter.split(':', 1)334 else:335 comparison_type = None336 if not self._query_processor.has_selector(parameter):337 continue338 for value in values: # forms keys can have multiple values339 queryset = self._query_processor.apply_selector(340 queryset, parameter, value,341 comparison_type=comparison_type)342 return queryset343 def _filtered_queryset(self):344 return self._apply_form_query(self._fresh_queryset())345 def get(self):346 queryset = self._filtered_queryset()347 items_per_page = self._read_int_parameter('items_per_page',348 self._DEFAULT_ITEMS_PER_PAGE)349 start_index = self._read_int_parameter('start_index', 0)350 page = queryset[start_index:(start_index + items_per_page)]351 rep = self._representation(page)352 rep.update({'total_results': len(queryset),353 'start_index': start_index,354 'items_per_page': items_per_page})355 return self._basic_response(rep)356 def full_representation(self):357 # careful, this rep can be huge for large collections358 return self._representation(self._fresh_queryset())359 def post(self):360 input_dict = self._decoded_input()361 try:362 instance = self.entry_class.create_instance(input_dict, self)363 entry = self._entry_from_instance(instance)364 entry.update(input_dict)365 except model_logic.ValidationError, exc:366 raise exceptions.BadRequest('Invalid input: %s' % exc)367 # RFC 2616 specifies that we provide the new URI in both the Location368 # header and the body369 response = http.HttpResponse(status=201, # Created370 content=entry.href())371 response['Location'] = entry.href()372 return response373class Relationship(Entry):374 _permitted_methods = ('GET', 'DELETE')375 # subclasses must override this with a dict mapping name to entry class376 related_classes = None377 def __init__(self, **kwargs):378 assert len(self.related_classes) == 2379 self.entries = dict((name, kwargs[name])380 for name in self.related_classes)381 for name in self.related_classes: # sanity check382 assert isinstance(self.entries[name], self.related_classes[name])383 # just grab the request from one of the entries384 some_entry = self.entries.itervalues().next()385 super(Relationship, self).__init__(some_entry._request)386 @classmethod387 def from_uri_args(cls, request, **kwargs):388 # kwargs contains URI args for each entry389 entries = {}390 for name, entry_class in cls.related_classes.iteritems():391 entries[name] = entry_class.from_uri_args(request, **kwargs)392 return cls(**entries)393 def _uri_args(self):394 kwargs = {}395 for name, entry in self.entries.iteritems():396 kwargs.update(entry._uri_args())397 return kwargs398 def short_representation(self):399 rep = self.link()400 for name, entry in self.entries.iteritems():401 rep[name] = entry.short_representation()402 return rep403 @classmethod404 def _get_related_manager(cls, instance):405 """Get the related objects manager for the given instance.406 The instance must be one of the related classes. This method will407 return the related manager from that instance to instances of the other408 related class.409 """410 this_model = type(instance)411 models = [entry_class.model for entry_class412 in cls.related_classes.values()]413 if isinstance(instance, models[0]):414 this_model, other_model = models415 else:416 other_model, this_model = models417 _, field = this_model.objects.determine_relationship(other_model)418 this_models_fields = (this_model._meta.fields419 + this_model._meta.many_to_many)420 if field in this_models_fields:421 manager_name = field.attname422 else:423 # related manager is on other_model, get name of reverse related424 # manager on this_model425 manager_name = field.related.get_accessor_name()426 return getattr(instance, manager_name)427 def _delete_entry(self):428 # choose order arbitrarily429 entry, other_entry = self.entries.itervalues()430 related_manager = self._get_related_manager(entry.instance)431 related_manager.remove(other_entry.instance)432 @classmethod433 def create_instance(cls, input_dict, containing_collection):434 other_name = containing_collection.unfixed_name435 cls._check_for_required_fields(input_dict, (other_name,))436 entry = containing_collection.fixed_entry437 other_entry = containing_collection.resolve_link(input_dict[other_name])438 related_manager = cls._get_related_manager(entry.instance)439 related_manager.add(other_entry.instance)440 return other_entry.instance441 def update(self, input_dict):442 pass443class RelationshipCollection(Collection):444 def __init__(self, request=None, fixed_entry=None):445 if request is None:446 request = fixed_entry._request447 super(RelationshipCollection, self).__init__(request)448 assert issubclass(self.entry_class, Relationship)449 self.related_classes = self.entry_class.related_classes450 self.fixed_name = None451 self.fixed_entry = None452 self.unfixed_name = None453 self.related_manager = None454 if fixed_entry is not None:455 self._set_fixed_entry(fixed_entry)456 entry_uri_arg = self.fixed_entry._uri_args().values()[0]457 self._query_params[self.fixed_name] = entry_uri_arg458 def _set_fixed_entry(self, entry):459 """Set the fixed entry for this collection.460 The entry must be an instance of one of the related entry classes. This461 method must be called before a relationship is used. It gets called462 either from the constructor (when collections are instantiated from463 other resource handling code) or from read_query_parameters() (when a464 request is made directly for the collection.465 """466 names = self.related_classes.keys()467 if isinstance(entry, self.related_classes[names[0]]):468 self.fixed_name, self.unfixed_name = names469 else:470 assert isinstance(entry, self.related_classes[names[1]])471 self.unfixed_name, self.fixed_name = names472 self.fixed_entry = entry473 self.unfixed_class = self.related_classes[self.unfixed_name]474 self.related_manager = self.entry_class._get_related_manager(475 entry.instance)476 def _query_parameters_accepted(self):477 return [(name, 'Show relationships for this %s' % entry_class.__name__)478 for name, entry_class479 in self.related_classes.iteritems()]480 def _resolve_query_param(self, name, uri_arg):481 entry_class = self.related_classes[name]482 return entry_class.from_uri_args(self._request, uri_arg)483 def read_query_parameters(self, query_params):484 super(RelationshipCollection, self).read_query_parameters(query_params)485 if not self._query_params:486 raise exceptions.BadRequest(487 'You must specify one of the parameters %s and %s'488 % tuple(self.related_classes.keys()))489 query_items = self._query_params.items()490 fixed_entry = self._resolve_query_param(*query_items[0])491 self._set_fixed_entry(fixed_entry)492 if len(query_items) > 1:493 other_fixed_entry = self._resolve_query_param(*query_items[1])494 self.related_manager = self.related_manager.filter(495 pk=other_fixed_entry.instance.id)496 def _entry_from_instance(self, instance):497 unfixed_entry = self.unfixed_class(self._request, instance)498 entries = {self.fixed_name: self.fixed_entry,499 self.unfixed_name: unfixed_entry}500 return self.entry_class(**entries)501 def _fresh_queryset(self):...

Full Screen

Full Screen

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run autotest automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful