CIRCL
/
lookyloo
mirror of https://github.com/CIRCL/lookyloo
2
0
Fork 0
Code Issues Releases Wiki Activity

chg: Properly type the response from LacusCore/PyLacus

pull/541/head
Raphaël Vinot 2022-10-26 14:25:23 +02:00
parent 73fb2fa652
commit e3075060cd
1 changed files with 3 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
bin/async_capture.py
View File

@ -8,8 +8,8 @@ from datetime import datetime
from pathlib import Path from pathlib import Path
from typing import Dict, Optional, Set, Union from typing import Dict, Optional, Set, Union
from lacuscore import LacusCore, CaptureStatus as CaptureStatusCore from lacuscore import LacusCore, CaptureStatus as CaptureStatusCore, CaptureResponse as CaptureResponseCore
from pylacus import PyLacus, CaptureStatus as CaptureStatusPy from pylacus import PyLacus, CaptureStatus as CaptureStatusPy, CaptureResponse as CaptureResponsePy
from redis import Redis from redis import Redis
from lookyloo.default import AbstractManager, get_config, get_socket_path, safe_create_dir from lookyloo.default import AbstractManager, get_config, get_socket_path, safe_create_dir
@ -62,6 +62,7 @@ class AsyncCapture(AbstractManager):
'''Process a query from the capture queue''' '''Process a query from the capture queue'''
self.set_running() self.set_running()
uuid: Optional[str] = None uuid: Optional[str] = None
entries: Union[CaptureResponseCore, CaptureResponsePy]
if isinstance(self.lacus, LacusCore): if isinstance(self.lacus, LacusCore):
if uuid := await self.lacus.consume_queue(): if uuid := await self.lacus.consume_queue():
entries = self.lacus.get_capture(uuid, decode=True) entries = self.lacus.get_capture(uuid, decode=True)