Returns a list of results from Twitter for the given query. - type : SEARCH, - start: Result.id or int, - count: maximum 100. There is a limit of 150+ queries per 15 minutes.
(self, query, type=SEARCH, start=1, count=10, sort=RELEVANCY, size=None, cached=False, **kwargs)
| 1430 | return url |
| 1431 | |
| 1432 | def search(self, query, type=SEARCH, start=1, count=10, sort=RELEVANCY, size=None, cached=False, **kwargs): |
| 1433 | """ Returns a list of results from Twitter for the given query. |
| 1434 | - type : SEARCH, |
| 1435 | - start: Result.id or int, |
| 1436 | - count: maximum 100. |
| 1437 | There is a limit of 150+ queries per 15 minutes. |
| 1438 | """ |
| 1439 | if type != SEARCH: |
| 1440 | raise SearchEngineTypeError |
| 1441 | if not query or count < 1 or (isinstance(start, (int, float)) and start < 1): |
| 1442 | return Results(TWITTER, query, type) |
| 1443 | if isinstance(start, (int, float)) and start < 10000: |
| 1444 | id = (query, kwargs.get("geo"), int(start), count) |
| 1445 | id = self._pagination.pop(id, "") |
| 1446 | else: |
| 1447 | id = start or "" |
| 1448 | # 1) Construct request URL. |
| 1449 | url = URL(TWITTER + "search/tweets.json?", method=GET) |
| 1450 | url.query = { |
| 1451 | "q": query, |
| 1452 | "max_id": id, |
| 1453 | "count": min(count, 100) |
| 1454 | } |
| 1455 | # 2) Restrict location with geo=(latitude, longitude, radius). |
| 1456 | # It can also be a (latitude, longitude)-tuple with default radius "10km". |
| 1457 | if "geo" in kwargs: |
| 1458 | url.query["geocode"] = ",".join((map(str, kwargs.pop("geo")) + ["10km"])[:3]) |
| 1459 | # 3) Restrict most recent with date="YYYY-MM-DD". |
| 1460 | # Only older tweets are returned. |
| 1461 | if "date" in kwargs: |
| 1462 | url.query["until"] = kwargs.pop("date") |
| 1463 | # 4) Restrict language. |
| 1464 | url.query["lang"] = self.language or "" |
| 1465 | # 5) Authenticate. |
| 1466 | url = self._authenticate(url) |
| 1467 | # 6) Parse JSON response. |
| 1468 | kwargs.setdefault("unicode", True) |
| 1469 | kwargs.setdefault("throttle", self.throttle) |
| 1470 | try: |
| 1471 | data = url.download(cached=cached, **kwargs) |
| 1472 | except HTTP420Error: |
| 1473 | raise SearchEngineLimitError |
| 1474 | except HTTP429TooMayRequests: |
| 1475 | raise SearchEngineLimitError |
| 1476 | data = json.loads(data) |
| 1477 | results = Results(TWITTER, query, type) |
| 1478 | results.total = None |
| 1479 | for x in data.get("statuses", []): |
| 1480 | r = Result(url=None) |
| 1481 | r.id = self.format(x.get("id_str")) |
| 1482 | r.url = self.format(TWITTER_STATUS % (x.get("user", {}).get("screen_name"), x.get("id_str"))) |
| 1483 | r.text = self.format(x.get("text")) |
| 1484 | r.date = self.format(x.get("created_at")) |
| 1485 | r.author = self.format(x.get("user", {}).get("screen_name")) |
| 1486 | r.profile = self.format(x.get("user", {}).get("profile_image_url")) # Profile picture URL. |
| 1487 | r.language = self.format(x.get("metadata", {}).get("iso_language_code")) |
| 1488 | results.append(r) |
| 1489 | # Twitter.search(start=id, count=10) takes a tweet.id, |
nothing calls this directly
no test coverage detected