Tail this function's logs. if keep_open, do so repeatedly, printing any new logs
(
self,
since,
filter_pattern,
limit=10000,
keep_open=True,
colorize=True,
http=False,
non_http=False,
force_colorize=False,
)
| 1329 | print("Done!") |
| 1330 | |
| 1331 | def tail( |
| 1332 | self, |
| 1333 | since, |
| 1334 | filter_pattern, |
| 1335 | limit=10000, |
| 1336 | keep_open=True, |
| 1337 | colorize=True, |
| 1338 | http=False, |
| 1339 | non_http=False, |
| 1340 | force_colorize=False, |
| 1341 | ): |
| 1342 | """ |
| 1343 | Tail this function's logs. |
| 1344 | if keep_open, do so repeatedly, printing any new logs |
| 1345 | """ |
| 1346 | |
| 1347 | try: |
| 1348 | since_stamp = string_to_timestamp(since) |
| 1349 | |
| 1350 | last_since = since_stamp |
| 1351 | while True: |
| 1352 | new_logs = self.zappa.fetch_logs( |
| 1353 | self.lambda_name, |
| 1354 | start_time=since_stamp, |
| 1355 | limit=limit, |
| 1356 | filter_pattern=filter_pattern, |
| 1357 | ) |
| 1358 | |
| 1359 | new_logs = [e for e in new_logs if e["timestamp"] > last_since] |
| 1360 | self.print_logs(new_logs, colorize, http, non_http, force_colorize) |
| 1361 | |
| 1362 | if not keep_open: |
| 1363 | break |
| 1364 | if new_logs: |
| 1365 | last_since = new_logs[-1]["timestamp"] |
| 1366 | time.sleep(1) |
| 1367 | except KeyboardInterrupt: # pragma: no cover |
| 1368 | # Die gracefully |
| 1369 | try: |
| 1370 | sys.exit(0) |
| 1371 | except SystemExit: |
| 1372 | os._exit(130) |
| 1373 | |
| 1374 | def undeploy(self, no_confirm=False, remove_logs=False): |
| 1375 | """ |