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_chunks(self):
self.assertEqual([i for i in utils.chunks([1, 2, 3], 1)], [[1], [2], [3]])
def _create_packages(
self, client, orders: list, package_type: OrderPackageType
) -> list:
packages = []
_package_cls = BetfairOrderPackage
limit = _package_cls.order_limit(package_type)
for chunked_orders in chunks(orders, limit):
order_package = _package_cls(
client=client,
market_id=self.market_id,
orders=chunked_orders,
package_type=package_type,
market=self.market,
)
packages.append(order_package)
orders.clear()
return packages
def poll_market_catalogue(context: dict, flumine) -> None:
client = flumine.client
live_markets = list(flumine.markets.markets.keys())
for market_ids in chunks(live_markets, 25):
try:
market_catalogues = client.betting_client.betting.list_market_catalogue(
filter=filters.market_filter(market_ids=market_ids),
max_results=25,
market_projection=[
"COMPETITION",
"EVENT",
"EVENT_TYPE",
"RUNNER_DESCRIPTION",
"RUNNER_METADATA",
"MARKET_START_TIME",
"MARKET_DESCRIPTION",
],
)
except BetfairError as e:
logger.error(