()
| 121 | shard_iterators: dict[str, str | None] = {} |
| 122 | |
| 123 | def refresh_shards(): |
| 124 | resp = streams_client.describe_stream(StreamArn=stream_arn) |
| 125 | for shard in resp["StreamDescription"]["Shards"]: |
| 126 | sid = shard["ShardId"] |
| 127 | if sid not in shard_iterators: |
| 128 | it = streams_client.get_shard_iterator( |
| 129 | StreamArn=stream_arn, |
| 130 | ShardId=sid, |
| 131 | ShardIteratorType="TRIM_HORIZON", |
| 132 | ) |
| 133 | shard_iterators[sid] = it["ShardIterator"] |
| 134 | log("POLLER", f"Tracking shard {sid}") |
| 135 | |
| 136 | refresh_shards() |
| 137 |
no test coverage detected