How to add a step to the ingest/index pipeline

[leonkleinmann]

Hi everyone,

I'm currently trying to add a stage to the ingest pipeline of aleph but it doesn't work.
I created a service just like translate but I don't understand how I can make aleph use it's worker.

docker-compose.yml

  ai:
    image: aleph-service:ai
    build:
      context: .
    volumes:
       - archive-data:/data
       - "./services/translate/translate:/translate/translate"
       - "/:/host"
    depends_on:
      - api
      - redis
    restart: on-failure
    environment:
      - REDIS_URL=redis://redis:6379/0
    env_file:
       - aleph.env

I also changed the settings.py

        # Document processing pipeline
        self.INGEST_PIPELINE = env.to_list(
            "ALEPH_INGEST_PIPELINE", [self.STAGE_ANALYZE, self.STAGE_AI]
        )

This surely is not enough to make Aleph use my AI stage.
Any help is welcome!

Thanks in Advance!

[stchris]

Hi @leonkleinmann,

I'm assuming you've seen the documentation at https://github.com/alephdata/translate-service and are trying to make your new service sort of work like the translation service in that example. Unfortunately that example is slightly outdated as of Aleph 4.0+, but the principles stay the same: you need to implement your own worker which inherits from servicelayer.worker.Worker. Specifically you need to focus on the dispatch_task method which is passing you a task ready for processing.

Other examples of these kinds of workers are ingest-file (https://github.com/alephdata/ingest-file/blob/3860024df29dc04d5f37dd47aee85f936f93c040/ingestors/worker.py#L28) and the aleph-worker (

aleph/aleph/worker.py

Line 109 in ef34516

class AlephWorker(Worker):

).

I hope this helps. If you have further questions please join us on the Aleph Discourse forum.

[leonkleinmann]

Hello @stchris

thank your for your reply.
Your hint to use aleph-ingest as "base" was very good and my worker is now consuming tasks.

Thank you very much!