From 9e205242d9dcb5f9fa096bb76ad6c06658766c8b Mon Sep 17 00:00:00 2001 From: Kevin Glisson Date: Mon, 6 Feb 2023 11:21:23 -0800 Subject: [PATCH 1/3] Fixes signal schema --- src/dispatch/signal/flows.py | 4 ++++ src/dispatch/signal/models.py | 13 ++++++++++--- src/dispatch/signal/scheduled.py | 24 ++++++++++++++---------- 3 files changed, 28 insertions(+), 13 deletions(-) diff --git a/src/dispatch/signal/flows.py b/src/dispatch/signal/flows.py index ed0d78603869..ba7d80b18429 100644 --- a/src/dispatch/signal/flows.py +++ b/src/dispatch/signal/flows.py @@ -4,6 +4,7 @@ from dispatch.case import service as case_service from dispatch.case import flows as case_flows from dispatch.signal import service as signal_service +from dispatch.tag import service as tag_service from dispatch.signal.models import SignalInstanceCreate, RawSignal @@ -27,6 +28,9 @@ def create_signal_instance( db_session=db_session, signal_instance_in=signal_instance_in ) + # associate any known tags with the signal + tag = tag_service.get_by_name(db_session=db_session, project_id=project.id, name="foo") + signal_instance.signal = signal db_session.commit() diff --git a/src/dispatch/signal/models.py b/src/dispatch/signal/models.py index 82ad14ef2988..5d5e69c54e68 100644 --- a/src/dispatch/signal/models.py +++ b/src/dispatch/signal/models.py @@ -215,14 +215,21 @@ class SignalPagination(DispatchBase): total: int +class AdditionalMetadata(DispatchBase): + name: Optional[str] + value: Optional[str] + type: Optional[str] + important: Optional[bool] + + class RawSignal(DispatchBase): action: Optional[List[Dict]] = [] - additional_metadata: Optional[List[Dict]] = Field([]) + additional_metadata: Optional[List[AdditionalMetadata]] = Field([], alias="additionalMetadata") asset: Optional[List[Dict]] = [] identity: Optional[Dict] = {} - origin_location: Optional[List[Dict]] = Field([]) + origin_location: Optional[List[Dict]] = Field([], alias="originLocation") variant: Optional[str] = None - created_at: Optional[datetime] = Field(None) + created_at: Optional[datetime] = Field(None, fields="createdAt") id: Optional[str] diff --git a/src/dispatch/signal/scheduled.py b/src/dispatch/signal/scheduled.py index 2a2317f723a1..6b4bf62073d4 100644 --- a/src/dispatch/signal/scheduled.py +++ b/src/dispatch/signal/scheduled.py @@ -35,14 +35,18 @@ def consume_signals(db_session: SessionLocal, project: Project): for plugin in plugins: log.debug(f"Consuming signals. Signal Consumer: {plugin.plugin.slug}") signal_instances = plugin.instance.consume() + if plugin.plugin.slug == "det-eng-sqs-signal-consumer": + with open("data.json", "w") as f: + for signal_instance_data in signal_instances: + log.debug(signal_instance_data.json()) + print(signal_instance_data.json(), file=f) - for signal_instance_data in signal_instances: - try: - signal_flows.create_signal_instance( - db_session=db_session, - project=project, - signal_instance_data=signal_instance_data, - ) - except Exception as e: - log.debug(signal_instance_data) - log.exception(e) + # try: + # signal_flows.create_signal_instance( + # db_session=db_session, + # project=project, + # signal_instance_data=signal_instance_data, + # ) + # except Exception as e: + # log.debug(signal_instance_data) + # log.exception(e) From 7911735d52f6ff6ed75da5f6b5c54f8284afcd35 Mon Sep 17 00:00:00 2001 From: Kevin Glisson Date: Mon, 6 Feb 2023 14:13:14 -0800 Subject: [PATCH 2/3] Fixing scheduler --- src/dispatch/signal/scheduled.py | 27 +++++++++++---------------- 1 file changed, 11 insertions(+), 16 deletions(-) diff --git a/src/dispatch/signal/scheduled.py b/src/dispatch/signal/scheduled.py index 6b4bf62073d4..400545da86f2 100644 --- a/src/dispatch/signal/scheduled.py +++ b/src/dispatch/signal/scheduled.py @@ -17,7 +17,7 @@ log = logging.getLogger(__name__) -# TODO do we want per signal source flexibility? + TODO do we want per signal source flexibility? @scheduler.add(every(1).minutes, name="signal-consume") @scheduled_project_task def consume_signals(db_session: SessionLocal, project: Project): @@ -35,18 +35,13 @@ def consume_signals(db_session: SessionLocal, project: Project): for plugin in plugins: log.debug(f"Consuming signals. Signal Consumer: {plugin.plugin.slug}") signal_instances = plugin.instance.consume() - if plugin.plugin.slug == "det-eng-sqs-signal-consumer": - with open("data.json", "w") as f: - for signal_instance_data in signal_instances: - log.debug(signal_instance_data.json()) - print(signal_instance_data.json(), file=f) - - # try: - # signal_flows.create_signal_instance( - # db_session=db_session, - # project=project, - # signal_instance_data=signal_instance_data, - # ) - # except Exception as e: - # log.debug(signal_instance_data) - # log.exception(e) + for signal_instance_data in signal_instances: + try: + signal_flows.create_signal_instance( + db_session=db_session, + project=project, + signal_instance_data=signal_instance_data, + ) + except Exception as e: + log.debug(signal_instance_data) + log.exception(e) From 632c9598eff23144fbcd630633d1e22ec1b101e0 Mon Sep 17 00:00:00 2001 From: Kevin Glisson Date: Tue, 7 Feb 2023 08:21:04 -0800 Subject: [PATCH 3/3] Fix comment --- src/dispatch/signal/scheduled.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/dispatch/signal/scheduled.py b/src/dispatch/signal/scheduled.py index 400545da86f2..1604fd8b206e 100644 --- a/src/dispatch/signal/scheduled.py +++ b/src/dispatch/signal/scheduled.py @@ -17,7 +17,7 @@ log = logging.getLogger(__name__) - TODO do we want per signal source flexibility? +# TODO do we want per signal source flexibility? @scheduler.add(every(1).minutes, name="signal-consume") @scheduled_project_task def consume_signals(db_session: SessionLocal, project: Project):