KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _update_query_params method in gabbi

Best Python code snippet using gabbi_python

queryset.py

Source:queryset.py

...20        """Update query with default_query if any."""21        return (query22                if not self.default_query23                else {'$and': [self.default_query, query]})24    def _update_query_params(self, params):25        """Update additional params to query."""26        if (self.session is not None) and ('session' not in params):27            params['session'] = self.session28        return params29    def clone(self):30        """Return a copy of queryset."""31        qs = type(self)(self.doc_class, self.db, session=self.session)32        qs.default_query = self.default_query33        qs.collection = self.collection34        return qs35    async def create(self, **kwargs):36        """Create document.37        Args:38            **kwargs: fields of the document.39        Returns:40            Document instance.41        """42        obj = self.doc_class(**kwargs)43        await obj.save(db=self.db, session=self.session)44        return obj45    async def create_indexes(self):46        """Create document's indexes defined in Meta class."""47        if self.doc_class.meta.indexes:48            await self.collection.create_indexes(self.doc_class.meta.indexes)49    async def delete_one(self, query, **kwargs):50        """Delete one document."""51        res = await self.collection.delete_one(52            self._update_query(query),53            **self._update_query_params(kwargs))54        return res.deleted_count if res.acknowledged else None55    async def delete_many(self, query, **kwargs):56        """Delete many documents."""57        res = await self.collection.delete_many(58            self._update_query(query),59            **self._update_query_params(kwargs))60        return res.deleted_count if res.acknowledged else None61    async def replace_one(self, query, *args, **kwargs):62        """Replace one document.63        Returns:64            int: Number of modified documents.65        Raises:66            aiomongodel.errors.DuplicateKeyError: On duplicate key error.67        """68        try:69            res = await self.collection.replace_one(70                self._update_query(query),71                *args,72                **self._update_query_params(kwargs))73        except pymongo.errors.DuplicateKeyError as e:74            raise DuplicateKeyError(str(e)) from e75        return res.modified_count if res.acknowledged else None76    async def update_one(self, query, *args, **kwargs):77        """Update one document.78        Returns:79            int: Number of modified documents.80        Raises:81            aiomongodel.errors.DuplicateKeyError: On duplicate key error.82        """83        try:84            res = await self.collection.update_one(85                self._update_query(query),86                *args,87                **self._update_query_params(kwargs))88        except pymongo.errors.DuplicateKeyError as e:89            raise DuplicateKeyError(str(e)) from e90        return res.modified_count if res.acknowledged else None91    async def update_many(self, query, *args, **kwargs):92        """Update many documents.93        Returns:94            int: Number of modified documents.95        Raises:96            aiomongodel.errors.DuplicateKeyError: On duplicate key error.97        """98        try:99            res = await self.collection.update_many(100                self._update_query(query),101                *args,102                **self._update_query_params(kwargs))103        except pymongo.errors.DuplicateKeyError as e:104            raise DuplicateKeyError(str(e)) from e105        return res.modified_count if res.acknowledged else None106    async def insert_one(self, *args, **kwargs):107        """Insert one document.108        Returns:109            Inserted ``_id`` value.110        Raises:111            aiomongodel.errors.DuplicateKeyError: On duplicate key error.112        """113        try:114            res = await self.collection.insert_one(115                *args,116                **self._update_query_params(kwargs))117        except pymongo.errors.DuplicateKeyError as e:118            raise DuplicateKeyError(str(e)) from e119        return res.inserted_id if res.acknowledged else None120    async def insert_many(self, *args, **kwargs):121        """Insert many documents.122        Returns:123            list: List of inserted ``_id`` values.124        """125        res = await self.collection.insert_many(126            *args,127            **self._update_query_params(kwargs))128        return res.inserted_ids if res.acknowledged else None129    async def find_one(self, query={}, *args, **kwargs):130        """Find one document.131        Arguments are the same as for ``motor.Collection.find_one``.132        This method does not returns ``None`` if there is no documents for133        given query but raises ``aiomongodel.errors.DocumentNotFoundError``.134        Returns:135            Document model instance.136        Raises:137            aiomongodel.errors.DocumentNotFoundError: If there is no documents138                for given query.139        """140        data = await self.collection.find_one(141            self._update_query(query),142            *args,143            **self._update_query_params(kwargs))144        if data is None:145            raise DocumentNotFoundError()146        return self.doc_class.from_mongo(data)147    async def get(self, _id, *args, **kwargs):148        """Get document by its _id."""149        return await self.find_one(150            {'_id': self.doc_class._id.to_mongo(_id)},151            *args,152            **self._update_query_params(kwargs))153    def find(self, query={}, *args, sort=None, **kwargs):154        """Find documents by query.155        Returns:156            MotorQuerySetCursor: Cursor to get actual data.157        """158        if not sort and self.default_sort:159            sort = self.default_sort160        return MotorQuerySetCursor(161            self.doc_class,162            self.collection.find(163                self._update_query(query),164                *args,165                sort=sort,166                **self._update_query_params(kwargs)))167    async def count(self, query={}, **kwargs):168        """Count documents in collection."""169        warnings.warn("Use `count_documents` instead",170                      DeprecationWarning,171                      stacklevel=2)172        return await self.collection.count_documents(173            self._update_query(query),174            **self._update_query_params(kwargs))175    async def count_documents(self, query={}, **kwargs):176        """Count documents in collection."""177        return await self.collection.count_documents(178            self._update_query(query),179            **self._update_query_params(kwargs))180    def aggregate(self, pipeline, **kwargs):181        """Return Aggregation cursor."""182        if not self.default_query:183            return self.collection.aggregate(184                pipeline,185                **self._update_query_params(kwargs))186        try:187            match = pipeline[0]['$match']188        except KeyError:189            return self.collection.aggregate(190                [{'$match': self.default_query}] + pipeline,191                **self._update_query_params(kwargs))192        else:193            pipeline[0]['$match'] = self._update_query(match)194            return self.collection.aggregate(195                pipeline,196                **self._update_query_params(kwargs))197    def with_options(self, **kwargs):198        """Change collection options."""199        clone = self.clone()200        clone.collection = self.collection.with_options(**kwargs)201        return clone202class MotorQuerySetCursor(object):203    """Cursor based on motor cursor."""204    DIRECT_TO_MOTOR = {'distinct', 'explain'}205    def __init__(self, doc_class, cursor):206        self.doc_class = doc_class207        self.cursor = cursor208    def _proxy_to_motor_cursor(self, method, *args, **kwargs):209        getattr(self.cursor, method)(*args, *kwargs)210        return self...

url_operations.py

Source:url_operations.py

1import urllib2import cgi3import urlparse4def get_path(url):5    scheme, host, path, query, fragment = urlparse.urlsplit(url)6    return path7def get_host(url):8    scheme, host, path, query, fragment = urlparse.urlsplit(url)9    return host10def add_path(url, new_path):11    """Given a url and path, return a new url that combines12    the two.13    """14    scheme, host, path, query, fragment = urlparse.urlsplit(url)15    new_path = new_path.lstrip('/')16    if path.endswith('/'):17        path += new_path18    else:19        path += '/' + new_path20    return urlparse.urlunsplit([scheme, host, path, query, fragment])21def _query_param(key, value):22    """ensure that a query parameter's value is a string23    of bytes in UTF-8 encoding.24    """25    if isinstance(value, unicode):26        pass27    elif isinstance(value, str):28        value = value.decode('utf-8')29    else:30        value = unicode(value)31    return (key, value.encode('utf-8'))32def _make_query_tuples(params):33    if hasattr(params, 'items'):34        return [_query_param(*param) for param in params.items()]35    else:36        return [_query_param(*params)]37def add_query_params(url, params):38    """use the _update_query_params function to set a new query39    string for the url based on params.40    """41    return update_query_params(url, params, update=False)42def update_query_params(url, params, update=True):43    """Given a url and a tuple or dict of parameters, return44    a url that includes the parameters as a properly formatted45    query string.46    If update is True, change any existing values to new values47    given in params.48    """49    scheme, host, path, query, fragment = urlparse.urlsplit(url)50    # cgi.parse_qsl gives back url-decoded byte strings. Leave these as51    # they are: they will be re-urlencoded below52    query_bits = [(k, v) for k, v in cgi.parse_qsl(query)]53    if update:54        query_bits = dict(query_bits)55        query_bits.update(_make_query_tuples(params))56    else:57        query_bits.extend(_make_query_tuples(params))58    query = urllib.urlencode(query_bits)...

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.