Assemblyline 4 Service Base
raw JSON → 4.7.3.1 verified Mon Apr 27 auth: no python
Base library for building services in Assemblyline 4, a scalable file analysis framework. Current version 4.7.3.1. Releases are frequent with a dev release cadence.
pip install assemblyline-v4-service Common errors
error ModuleNotFoundError: No module named 'assemblyline' ↓
cause Installing the wrong package (assemblyline-v4-service is correct) or missing dependency.
fix
pip install assemblyline-v4-service
error ImportError: cannot import name 'ServiceBase' from 'assemblyline_v4_service' ↓
cause Using wrong import path; correct path is assemblyline_v4_service.common.base.
fix
from assemblyline_v4_service.common.base import ServiceBase
error AttributeError: 'dict' object has no attribute 'get_param' ↓
cause execute() received a dict instead of a ServiceRequest object.
fix
Change execute method signature to def execute(self, request: ServiceRequest):
Warnings
breaking Namespace changed from assemblyline to assemblyline_v4_service in v4.0.0. All imports must be updated. ↓
fix Replace 'from assemblyline.common' with 'from assemblyline_v4_service.common'.
gotcha ServiceBase.execute() now expects a ServiceRequest object, not a dict. Passing a dict will cause AttributeError. ↓
fix Define execute(self, request: ServiceRequest) and use request.get_param(), etc.
gotcha Must call self.start() in __main__ block; otherwise service won't initialize correctly. ↓
fix Add if __name__ == '__main__': MyService().start()
Imports
- ServiceBase wrong
from assemblyline.common import ServiceBasecorrectfrom assemblyline_v4_service.common.base import ServiceBase - Task wrong
from assemblyline.common.task import Taskcorrectfrom assemblyline_v4_service.common.task import Task
Quickstart
import os
from assemblyline_v4_service.common.base import ServiceBase
from assemblyline_v4_service.common.request import ServiceRequest
from assemblyline_v4_service.common.result import Result
class MyService(ServiceBase):
def start(self):
self.log.info(f'Starting service, API_KEY={os.environ.get("API_KEY", "")[:4]}...')
def execute(self, request: ServiceRequest) -> None:
result = Result()
request.result = result
if __name__ == '__main__':
MyService().start()