Repository: spicyparrot/kafka_scrapy_connect (language: python)

Issue:
feat: create dependabot action

feat: create dependabot action



Discussion / hints:

Fix patch:
diff --git a/.github/pull_request_template.md b/.github/pull_request_template.md
new file mode 100644
index 0000000..750430a
--- /dev/null
+++ b/.github/pull_request_template.md
@@ -0,0 +1,5 @@
+# Changes
+
+# Checklist
+- [ ] Tests
+- [ ] Documentation
\ No newline at end of file
diff --git a/.github/workflows/create_release.yml b/.github/workflows/create_release.yml
new file mode 100644
index 0000000..32fde8b
--- /dev/null
+++ b/.github/workflows/create_release.yml
@@ -0,0 +1,67 @@
+name: 📢 Create Release
+on:
+  workflow_dispatch:
+    inputs:
+      bump:
+        type: choice
+        description: 🌍 Release bump
+        options:
+          - patch
+          - minor
+          - major
+
+concurrency:
+  group: create_release
+  cancel-in-progress: true
+
+env:
+  RED: '\033[0;31m'
+  GREEN: '\033[0;32m'
+  YELLOW: '\033[0;33m'
+  PURPLE: '\033[0;35m'
+  WHITE: '\033[0m'
+
+jobs:
+  #test:
+  #  name: 🔬 Test Action
+  #  runs-on: ubuntu-latest
+  #  steps:
+  #    - uses: actions/checkout@v3
+  #
+  #    - name: Check Dependabot Alerts
+  #      id: alerts
+  #      uses: spicyparrot/check-dependabot@trunk              # Check out yourself to test
+  #      with:
+  #        github_personal_token: ${{ secrets.ACTIONS_ACCESS_TOKEN }}
+
+  create_release:
+    name: 📢 Create Release
+    runs-on: ubuntu-latest
+    #needs: [test]
+    steps:
+      - name: 🛀 Checkout Self
+        uses: actions/checkout@v4.1.1
+
+      - name: 🏷 Create Tag
+        id: tag
+        uses: mathieudutour/github-tag-action@v6.1 #https://github.com/marketplace/actions/github-tag
+        with:
+          github_token: ${{ github.token }}
+          tag_prefix: "v"
+          default_bump: ${{ github.event.inputs.bump }}
+          release_branches: trunk
+
+      - name: 📢 GitHub Release
+        uses: ncipollo/release-action@v1.13.0
+        with:
+          tag: "${{ steps.tag.outputs.new_tag }}"
+          token: ${{ github.token }}
+          generateReleaseNotes: true
+          allowUpdates: true
+
+      - name: 📝 Details
+        shell: bash
+        run: |
+          echo "Release Type = ${{ steps.tag.outputs.release_type }}" >> $GITHUB_STEP_SUMMARY
+          echo "New Tag = ${{ steps.tag.outputs.new_tag }}" >> $GITHUB_STEP_SUMMARY
+          echo "Previous Tag = ${{ steps.tag.outputs.previous_tag  }}" >> $GITHUB_STEP_SUMMARY
diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..6769e21
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,160 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# poetry
+#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
+#poetry.lock
+
+# pdm
+#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
+#pdm.lock
+#   pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
+#   in version control.
+#   https://pdm.fming.dev/#use-with-ide
+.pdm.toml
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+
+# PyCharm
+#  JetBrains specific template is maintained in a separate JetBrains.gitignore that can
+#  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
+#  and can be added to the global gitignore or merged into this file.  For a more nuclear
+#  option (not recommended) you can uncomment the following to ignore the entire idea folder.
+#.idea/
\ No newline at end of file
diff --git a/README.md b/README.md
index 2eead23..46faf33 100644
--- a/README.md
+++ b/README.md
@@ -1,2 +1,122 @@
-# kafka_scrapy_connect
-kafka_scrapy_connect is a custom Scrapy library that aims to integrates Scrapy with Kafka.
+[![Python 3.9](https://img.shields.io/badge/python-3.9-blue.svg)](https://www.python.org/downloads/release/python-3918) [![Python 3.10](https://img.shields.io/badge/python-3.10-blue.svg)](https://www.python.org/downloads/release/python-31013/) [![Python 3.11](https://img.shields.io/badge/python-3.11-blue.svg)](https://www.python.org/downloads/release/python-3117/)
+## 🚀 Kafka Scrapy Connect
+
+### Overview
+
+`kafka_scrapy_connect` is a custom Scrapy library that aims to integrates Scrapy with Kafka.
+
+It consists of two main components: spiders and pipelines, which interact with Kafka for message consumption and item publishing.
+
+This project has been motivated by the great work undertaken in: https://github.com/dfdeshom/scrapy-kafka. 
+
+`kafka_scrapy_connect` utilises [Confluent's](https://github.com/confluentinc/confluent-kafka-python) Kafka Python client under the hood, to provide high-level producer and consumer features.
+## Features
+
+1️⃣ **Integration with Kafka** 📈
+  - Enables communication between Scrapy spiders and Kafka topics for efficient data processing.
+  - Through partitions and consumer groups message processing can be parallelised across multiple spiders!
+  - Reduces overhead and improves throughput by giving the user the ability to consume messages in batches.
+
+2️⃣ **Customisable Settings** 🛠️
+  - Provides flexibility through customisable configuration for both consumers and producers.
+
+3️⃣ **Error Handling** 🚑
+  - Automatically handles network errors during crawling and publishes failed URLs to a designated output topic. 
+
+4️⃣ **Serialisation Customisation** 🧬
+  - Allows users to customize how Kafka messages are deserializsd by overriding the process_kafka_message method.
+
+## Installation
+
+You can install `kafka_scrapy_connect` via pip:
+```
+pip install kafka_scrapy_connect
+```
+
+## Example
+
+A full [example](https://github.com/spicyparrot/kafka_scrapy_connect.git) using the `kafka_scrapy_connect` library can be found inside the repo.
+
+The *only prerequisite for walking through the exampl*e is the installation of **Docker** 🐳 . 
+
+This is needed because a kafka cluster will be created locally using containers so `kafka_scrapy_connect` can communicate with a broker. 
+
+If all set, follow the below steps 👇
+
+1. Create a virtual environment, clone the repo and install requirements:
+```bash
+python3 -m venv .venv
+source .venv/bin/activate
+git clone https://github.com/spicyparrot/kafka_scrapy_connect.git && cd kafka_scrapy_connect
+pip install -r requirements.txt
+```
+2. Create a local kafka cluster with required topics:
+
+```bash
+bash ./examples/kafka/kafka_start.sh --input-topic ScrapyInput,1 --output-topic ScrapyOutput,1 --error-topic ScrapyError,1
+```
+
+3. Initiate the spider:
+```bash
+cd examples/quotes && scrapy crawl quotes
+```
+
+4. Publish a message to the input kafka topic and watch the spider consume and process the mesasge 🪄 
+   1. ☝️ This will require some custom producer code to publish messages or go to https://localhost:8080 and use the UI to publish some example messages in! (*The UI was created when bringing up your local kafka cluster*)
+
+
+5. When satisfied with testing, exit the spider and clean up the local kafka cluster:
+```bash
+bash ./examples/kafka/kafka_stop.sh
+```
+
+## Usage
+
+### Custom Settings
+`kafka_scrapy_connect` supports the following custom settings:
+
+- `SCRAPY_KAFKA_HOSTS`  - A list of kafka broker hosts. (Default: `localhost:29092`)
+- `SCRAPY_INPUT_TOPIC`  - Topic from which the spider[s] *consumes* messages from. (Default: `ScrapyInput`)
+- `SCRAPY_OUTPUT_TOPIC` - Topic where scraped items are published. (Default: `ScrapyOutput`)
+- `SCRAPY_ERROR_TOPIC`  - Topic for publishing URLs that failed due to *network errors*. (Default: `ScrapyError`)
+- `SCRAPY_CONSUMER_CONFIG` - Additional configuration options for Kafka consumers (see [here](https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md))
+- `SCRAPY_PRODUCER_CONFIG` - Additional configuration options for Kafka producers (see [here](https://github.com/confluentinc/librdkafka/blob/master/CONFIGURATION.md))
+- `SCRAPY_KAFKA_PRODUCER` - Key used for partitioning messages in Kafka producer (Default: `""` *Roundrobin*)
+- `SCRAPY_KAFKA_PRODUCER_CALLBACKS` - Enable or disable asynchronous message delivery callbacks. (Default: `False`)
+
+### Customisation
+
+**Customising deserialisation** 
+
+You can customize how Kafka messages are deserialized by overriding the process_kafka_message method in your spider class. 
+
+This allows for handling custom message formats or data transformations.
+
+```python
+class CustomSpider(KafkaListeningSpider):
+    def process_kafka_message(self, message, meta={}, headers={}):
+        # Custom deserialization logic
+        # Return URL, metadata or None if extraction fails
+        pass
+```
+
+⚠️ *By default*, if no custom `process_kafka_message` method is provided, the spider method `process_kafka_message` **will expect** a JSON payload or a string containing a valid url. If it's a JSON object, it expects `url` in the K/V pair.
+
+**Customising Producer & Consumer settings**
+
+You can customize producer and consumer settings by providing a dictionary of configuration options in your Scrapy settings under `SCRAPY_PRODUCER_CONFIG` and `SCRAPY_CONSUMER_CONFIG`.
+
+```python
+# Example SCRAPY_PRODUCER_CONFIG
+SCRAPY_PRODUCER_CONFIG = {
+    'compression.type': 'gzip',
+    'request.timeout.ms': 5000
+}
+
+# Example SCRAPY_CONSUMER_CONFIG
+SCRAPY_CONSUMER_CONFIG = {
+    'fetch.wait.max.ms': 10,
+    'max.poll.interval.ms': 600000,
+    'auto.offset.reset': 'latest'
+}
+```
\ No newline at end of file
diff --git a/examples/kafka/docker-compose.yml b/examples/kafka/docker-compose.yml
new file mode 100644
index 0000000..b527d17
--- /dev/null
+++ b/examples/kafka/docker-compose.yml
@@ -0,0 +1,34 @@
+version: '2'
+services:
+  zookeeper:
+    image: confluentinc/cp-zookeeper:latest
+    container_name: zookeeper
+    environment:
+      ZOOKEEPER_CLIENT_PORT: 2181
+      ZOOKEEPER_TICK_TIME: 2000
+    ports:
+      - 22181:2181
+  
+  kafka:
+    image: confluentinc/cp-kafka:latest
+    container_name: kafka
+    depends_on:
+      - zookeeper
+    ports:
+      - 29092:29092
+    environment:
+      KAFKA_BROKER_ID: 1
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
+      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+
+  kafka-ui:
+    container_name: kafka-ui
+    image: provectuslabs/kafka-ui:master
+    ports:
+      - 8080:8080
+    environment:
+      KAFKA_CLUSTERS_0_NAME: testing_cluster
+      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092
\ No newline at end of file
diff --git a/examples/kafka/kafka_start.sh b/examples/kafka/kafka_start.sh
new file mode 100755
index 0000000..9f7399a
--- /dev/null
+++ b/examples/kafka/kafka_start.sh
@@ -0,0 +1,80 @@
+#!/bin/bash
+
+# Constants
+SCRIPT_DIR="$(cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null && pwd)"
+
+cd $SCRIPT_DIR
+
+# Load in logging library
+. "${SCRIPT_DIR}/../utilities/log.sh"
+
+# Extract kafka version and topics from cmd line args
+while [[ $# -gt 0 ]]; do
+    case $1 in
+        --input-topic)
+            INPUT_TOPIC="$2"
+            shift
+            shift
+            ;;
+        --output-topic)
+            OUTPUT_TOPIC="$2"
+            shift
+            shift
+            ;;
+        --error-topic)
+            ERROR_TOPIC="$2"
+            shift
+            shift
+            ;;
+        *)
+            error "Invalid argument: $1"
+            exit 1
+            ;;
+    esac
+done
+
+# Check that the arguments were passed correctly
+if [[ -z $INPUT_TOPIC || -z $OUTPUT_TOPIC || -z $ERROR_TOPIC ]]; then
+    error "Can't execute script"
+    warn "Usage: $0 --input-topic inputTopic,1 --output-topic outputTopic,1 --error-topic errorTopic,1"
+    exit 1
+fi
+
+# Function to validate topic format
+validate_topic_format() {
+    local topic="$1"
+    if [[ ! "$topic" =~ ^[a-zA-Z0-9_-]+,[0-9]+$ ]]; then
+        error "Invalid format for topic: $topic"
+        warn "Usage: $0 --input-topic <topic_name>,<partitions> --output-topic <topic_name>,<partitions> --error-topic <topic_name>,<partitions>"
+        exit 1
+    fi
+}
+
+# Validate topic formats
+validate_topic_format "$INPUT_TOPIC"
+validate_topic_format "$OUTPUT_TOPIC"
+validate_topic_format "$ERROR_TOPIC"
+
+# Concatenate topics into one variable
+TOPICS="$INPUT_TOPIC,$OUTPUT_TOPIC,$ERROR_TOPIC"
+
+
+info "Bringing up local kafka cluster"
+docker-compose up -d
+sleep 10 
+
+# Log information about connecting to Kafka
+info "Please connect to the following localhost:29092 to access Kafka"
+
+# Split the topics string into an array
+IFS=',' read -ra TOPIC_PARTITIONS <<< "$TOPICS"
+
+# Print the parsed topic-partition pairs for verification
+for ((i=0; i<${#TOPIC_PARTITIONS[@]}; i+=2)); do
+    TOPIC="${TOPIC_PARTITIONS[i]}"
+    PARTITION="${TOPIC_PARTITIONS[i+1]}"
+    info "Creating Kafka topic $TOPIC with $PARTITION partition[s]"
+done
+
+
+./scripts/create_topics.sh --topics $TOPICS
\ No newline at end of file
diff --git a/examples/kafka/kafka_stop.sh b/examples/kafka/kafka_stop.sh
new file mode 100755
index 0000000..bbef457
--- /dev/null
+++ b/examples/kafka/kafka_stop.sh
@@ -0,0 +1,17 @@
+#!/bin/bash
+
+# Constants
+SCRIPT_DIR="$(cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null && pwd)"
+
+cd $SCRIPT_DIR
+
+# Load in logging library
+. "${SCRIPT_DIR}/../utilities/log.sh"
+
+warn "Bringing down local kafka cluster 🫳"
+docker-compose down
+sleep 5
+
+# Log information about connecting to Kafka
+warn "Nuking images ☢️"
+docker system prune -a -f
diff --git a/examples/kafka/scripts/create_topics.sh b/examples/kafka/scripts/create_topics.sh
new file mode 100755
index 0000000..53d40ca
--- /dev/null
+++ b/examples/kafka/scripts/create_topics.sh
@@ -0,0 +1,43 @@
+#!/bin/bash
+# Script directory
+SCRIPT_DIR="$(cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null && pwd)"
+
+# Load in logging library
+. "${SCRIPT_DIR}/../../utilities/log.sh"
+
+# Parse command line arguments
+while [[ $# -gt 0 ]];do
+    case $1 in
+        --topics)
+            KAFKA_TOPIC_LIST="$2"
+            shift
+            shift
+            ;;
+        *)
+            error "Invalid argument: $1"
+            exit 1
+            ;;
+    esac
+done
+
+# Check that the arguments were passed correctly
+if [[ -z $KAFKA_TOPIC_LIST ]]; then
+    warn "Usage: $0 --topics foobar,1,test,3"
+    exit 1
+fi
+
+if [ -n "$KAFKA_TOPIC_LIST" ]; then
+  IFS=',' read -ra TOPICS <<< "$KAFKA_TOPIC_LIST"
+
+  DOCKER_COMMAND="docker exec kafka bash -c \""
+
+  for (( i=0; i<${#TOPICS[@]}; i+=2 )); do
+    topic=${TOPICS[i]}
+    partitions=${TOPICS[i+1]}
+    DOCKER_COMMAND+=" kafka-topics --create --bootstrap-server localhost:9092 --topic $topic --partitions $partitions;"
+  done
+  DOCKER_COMMAND+="\" "
+  eval $DOCKER_COMMAND
+else
+  info "No kafka topics to create."
+fi
\ No newline at end of file
diff --git a/examples/quotes/quotes/__init__.py b/examples/quotes/quotes/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/examples/quotes/quotes/items.py b/examples/quotes/quotes/items.py
new file mode 100644
index 0000000..47bd61c
--- /dev/null
+++ b/examples/quotes/quotes/items.py
@@ -0,0 +1,12 @@
+# Define here the models for your scraped items
+#
+# See documentation in:
+# https://docs.scrapy.org/en/latest/topics/items.html
+
+import scrapy
+
+
+class QuotesItem(scrapy.Item):
+    # define the fields for your item here like:
+    # name = scrapy.Field()
+    pass
diff --git a/examples/quotes/quotes/middlewares.py b/examples/quotes/quotes/middlewares.py
new file mode 100644
index 0000000..a90cfa6
--- /dev/null
+++ b/examples/quotes/quotes/middlewares.py
@@ -0,0 +1,103 @@
+# Define here the models for your spider middleware
+#
+# See documentation in:
+# https://docs.scrapy.org/en/latest/topics/spider-middleware.html
+
+from scrapy import signals
+
+# useful for handling different item types with a single interface
+from itemadapter import is_item, ItemAdapter
+
+
+class QuotesSpiderMiddleware:
+    # Not all methods need to be defined. If a method is not defined,
+    # scrapy acts as if the spider middleware does not modify the
+    # passed objects.
+
+    @classmethod
+    def from_crawler(cls, crawler):
+        # This method is used by Scrapy to create your spiders.
+        s = cls()
+        crawler.signals.connect(s.spider_opened, signal=signals.spider_opened)
+        return s
+
+    def process_spider_input(self, response, spider):
+        # Called for each response that goes through the spider
+        # middleware and into the spider.
+
+        # Should return None or raise an exception.
+        return None
+
+    def process_spider_output(self, response, result, spider):
+        # Called with the results returned from the Spider, after
+        # it has processed the response.
+
+        # Must return an iterable of Request, or item objects.
+        for i in result:
+            yield i
+
+    def process_spider_exception(self, response, exception, spider):
+        # Called when a spider or process_spider_input() method
+        # (from other spider middleware) raises an exception.
+
+        # Should return either None or an iterable of Request or item objects.
+        pass
+
+    def process_start_requests(self, start_requests, spider):
+        # Called with the start requests of the spider, and works
+        # similarly to the process_spider_output() method, except
+        # that it doesn’t have a response associated.
+
+        # Must return only requests (not items).
+        for r in start_requests:
+            yield r
+
+    def spider_opened(self, spider):
+        spider.logger.info("Spider opened: %s" % spider.name)
+
+
+class QuotesDownloaderMiddleware:
+    # Not all methods need to be defined. If a method is not defined,
+    # scrapy acts as if the downloader middleware does not modify the
+    # passed objects.
+
+    @classmethod
+    def from_crawler(cls, crawler):
+        # This method is used by Scrapy to create your spiders.
+        s = cls()
+        crawler.signals.connect(s.spider_opened, signal=signals.spider_opened)
+        return s
+
+    def process_request(self, request, spider):
+        # Called for each request that goes through the downloader
+        # middleware.
+
+        # Must either:
+        # - return None: continue processing this request
+        # - or return a Response object
+        # - or return a Request object
+        # - or raise IgnoreRequest: process_exception() methods of
+        #   installed downloader middleware will be called
+        return None
+
+    def process_response(self, request, response, spider):
+        # Called with the response returned from the downloader.
+
+        # Must either;
+        # - return a Response object
+        # - return a Request object
+        # - or raise IgnoreRequest
+        return response
+
+    def process_exception(self, request, exception, spider):
+        # Called when a download handler or a process_request()
+        # (from other downloader middleware) raises an exception.
+
+        # Must either:
+        # - return None: continue processing this exception
+        # - return a Response object: stops process_exception() chain
+        # - return a Request object: stops process_exception() chain
+        pass
+
+    def spider_opened(self, spider):
+        spider.logger.info("Spider opened: %s" % spider.name)
diff --git a/examples/quotes/quotes/pipelines.py b/examples/quotes/quotes/pipelines.py
new file mode 100644
index 0000000..2abd941
--- /dev/null
+++ b/examples/quotes/quotes/pipelines.py
@@ -0,0 +1,13 @@
+# Define your item pipelines here
+#
+# Don't forget to add your pipeline to the ITEM_PIPELINES setting
+# See: https://docs.scrapy.org/en/latest/topics/item-pipeline.html
+
+
+# useful for handling different item types with a single interface
+from itemadapter import ItemAdapter
+
+
+class QuotesPipeline:
+    def process_item(self, item, spider):
+        return item
diff --git a/examples/quotes/quotes/settings.py b/examples/quotes/quotes/settings.py
new file mode 100644
index 0000000..8bbbc6a
--- /dev/null
+++ b/examples/quotes/quotes/settings.py
@@ -0,0 +1,34 @@
+import os
+import sys
+import logging
+from scrapy.utils.log import configure_logging
+
+BOT_NAME = "quotes"
+
+SPIDER_MODULES = ["quotes.spiders"]
+NEWSPIDER_MODULE = "quotes.spiders"
+
+ITEM_PIPELINES = {
+    'kafka_scrapy_connect.pipelines.KafkaPublishPipeline': 100
+    }
+
+# Scrapy kafka connect settings
+SCRAPY_KAFKA_HOSTS  = 'localhost:29092'
+SCRAPY_INPUT_TOPIC  = 'ScrapyInput'
+SCRAPY_OUTPUT_TOPIC = 'ScrapyOutput'
+SCRAPY_ERROR_TOPIC = 'ScrapyErrors'
+
+SCRAPY_KAFKA_PRODUCER_KEY = ''
+SCRAPY_KAFKA_PRODUCER_CALLBACKS = False
+SCRAPY_PRODUCER_CONFIG = {
+    'queue.buffering.max.ms' : 1,
+    'linger.ms' : 5
+}
+SCRAPY_CONSUMER_CONFIG = {
+    'fetch.wait.max.ms': 10,
+    'max.poll.interval.ms': 600000,
+    'auto.offset.reset': 'latest'
+}
+
+LOG_LEVEL = 'INFO'  # to only display errors
+LOG_FORMAT = '%(asctime)s - %(levelname)8s - %(message)s'
\ No newline at end of file
diff --git a/examples/quotes/quotes/spiders/__init__.py b/examples/quotes/quotes/spiders/__init__.py
new file mode 100644
index 0000000..ebd689a
--- /dev/null
+++ b/examples/quotes/quotes/spiders/__init__.py
@@ -0,0 +1,4 @@
+# This package will contain the spiders of your Scrapy project
+#
+# Please refer to the documentation for information on how to create and manage
+# your spiders.
diff --git a/examples/quotes/quotes/spiders/quotes_spider.py b/examples/quotes/quotes/spiders/quotes_spider.py
new file mode 100644
index 0000000..ea1eaba
--- /dev/null
+++ b/examples/quotes/quotes/spiders/quotes_spider.py
@@ -0,0 +1,31 @@
+import os
+import sys
+import json
+import logging
+
+SCRIPT_DIRECTORY = os.path.dirname(os.path.realpath(__file__))
+
+sys.path.insert(0, os.path.abspath(f'{SCRIPT_DIRECTORY}/../../../../'))
+
+from kafka_scrapy_connect.spiders import KafkaListeningSpider
+
+class TestKafkaSpider(KafkaListeningSpider):
+    name = "quotes"
+
+    # Override process_kafka_message to suit your serializer type or use default method which can parse JSON / string serialized messages
+    # Method needs to return a URL to crawl / scrape
+    # def process_kafka_message(self, message):
+    #     logging.info('Custom process kafka message - received message')
+    #     json_obj = json.loads(message.value())
+    #     url = json_obj['url']
+    #     return url
+
+
+    def parse(self, response):
+        logging.info(f'Received a response ...{response}')
+        for quote in response.xpath('//div[@class="quote"]'):
+            yield {
+                'text' : quote.xpath('./span[@class="text"]/text()').extract_first(),
+                'author' : quote.xpath('.//small[@class="author"]/text()').extract_first(),
+                'tags' : quote.xpath('.//div[@class="tags"]/a[@class="tag"]/text()').extract()
+            }
\ No newline at end of file
diff --git a/examples/quotes/scrapy.cfg b/examples/quotes/scrapy.cfg
new file mode 100644
index 0000000..0fd11ca
--- /dev/null
+++ b/examples/quotes/scrapy.cfg
@@ -0,0 +1,17 @@
+# Automatically created by: scrapy startproject
+#
+# For more information about the [deploy] section see:
+# https://scrapyd.readthedocs.io/en/latest/deploy.html
+
+[settings]
+default = quotes.settings
+
+[deploy]
+#url = http://localhost:6800/
+project = quotes
+
+
+# New maui coming, rematch ipdb, same source but using the new maui. 
+# Kickfire data (stable data)
+# Generated 2.2 millions IPDB - 2 million / 4 
+# If i change MAUI, dont want to get a new kickfire
\ No newline at end of file
diff --git a/examples/utilities/log.sh b/examples/utilities/log.sh
new file mode 100644
index 0000000..8353b4d
--- /dev/null
+++ b/examples/utilities/log.sh
@@ -0,0 +1,67 @@
+#!/bin/bash
+
+# ANSI colour palette
+RESET='\033[0m'
+GREEN='\033[38;5;2m'
+RED='\033[38;5;1m'
+YELLOW='\033[38;5;3m'
+MAGENTA='\033[38;5;5m'
+CYAN='\033[38;5;6m'
+
+# Log message to stderr
+log() {
+    printf "%b\n" "$*" >&2
+}
+
+logColoured() {
+    local colour="$1"
+    local message="$2"
+    local before="${message%%:*}" # Extracts text before the first ":"
+    local after="${message#*:}"  # Extracts text after the first ":"
+    echo -e "\n${before} ==> ${colour}${after}${RESET}"
+}
+
+logColouredWrapper() {
+    local colour="$1"
+    local prefix="$2"
+    logColoured "$colour" "$prefix: $*"
+}
+
+logGreen() {
+    logColouredWrapper "$GREEN" "INFO"
+}
+
+logYellow() {
+    logColouredWrapper "$YELLOW" "WARN"
+}
+
+logPurple() {
+    logColouredWrapper "$MAGENTA" "OTHER"
+}
+
+logCyan() {
+    logColouredWrapper "$CYAN" "OTHER"
+}
+
+# Log info message
+info() {
+    logColoured "$GREEN" "INFO: $*"
+}
+
+# Log warning message
+warn() {
+    logColoured "$YELLOW" "WARN: $*"
+}
+
+# Log error message
+error() {
+    logColoured "$RED" "ERROR: $*"
+}
+
+silence() {
+    if ${DEBUG_MODE:-false}; then
+        "$@"
+    else
+        "$@" >/dev/null 2>&1
+    fi
+}
\ No newline at end of file
diff --git a/kafka_scrapy_connect/__init__.py b/kafka_scrapy_connect/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/kafka_scrapy_connect/pipelines.py b/kafka_scrapy_connect/pipelines.py
new file mode 100644
index 0000000..4ebf760
--- /dev/null
+++ b/kafka_scrapy_connect/pipelines.py
@@ -0,0 +1,78 @@
+from confluent_kafka import Producer as KafkaProducer
+from scrapy.exporters import BaseItemExporter
+import logging
+import json
+
+class KafkaPublishPipeline:
+    """
+    Publishes a serialized item into a Kafka topic.
+    """
+
+    def __init__(self, producer, topic, key='', drcb=False):
+        """
+        Initializes the Kafka item publisher.
+
+        Args:
+            producer (KafkaProducer): The Kafka producer.
+            topic (str): The Kafka topic being used.
+        """
+        self.producer = producer
+        self.topic = topic
+        self.key = key
+        self.drcb = drcb
+
+    def process_item(self, item, spider):
+        """
+        Processes the item and publishes it to Kafka.
+
+        Args:
+            item: Item being processed.
+            spider: The current spider being used.
+        """
+        payload = dict(item)
+        if self.drcb:
+            # Produce is asynchronous, all it does is enqueue the message to an internal queue
+            # Adding flush after every produce effectively makes it synchronous
+            self.producer.produce(self.topic, key=self.key, partition=-1,value=json.dumps(payload), callback=self.delivery_callback)
+        else:
+            logging.debug(f'Publishing results to {self.topic}')
+            self.producer.produce(self.topic, key=self.key, partition=-1,value=json.dumps(payload))
+        self.producer.poll(0)
+        return item
+
+    def delivery_callback(self, err, msg):
+        if err is not None:
+            logging.error(f'Failed to deliver message {msg}')
+        else:
+            logging.info(f'Produced to {msg.topic()} [{msg.partition()}] @ {msg.offset()}')
+
+    @classmethod
+    def from_settings(cls, settings):
+        """
+        Initializes the Kafka item publisher from Scrapy settings.
+
+        Args:
+            settings: The current Scrapy settings.
+
+        Returns:
+            KafkaItemPublisher: An instance of KafkaItemPublisher.
+        """
+        kafka_hosts = settings.get('SCRAPY_KAFKA_HOSTS', 'localhost:9092')
+        producer_config = settings.get('SCRAPY_PRODUCER_CONFIG', {})
+        kafka_config = {'bootstrap.servers': kafka_hosts, **producer_config}
+        topic = settings.get('SCRAPY_OUTPUT_TOPIC', 'scrapy_kafka_item')
+        key = settings.get('SCRAPY_KAFKA_PRODUCER_KEY', '')
+        drcb = settings.get('SCRAPY_KAFKA_PRODUCER_CALLBACKS', False)
+        logging.info(f'Instantiated a kafka producer for topic: {topic} with the following configuration: {kafka_config}')
+        #kafka_config = {'bootstrap.servers': kafka_hosts}
+        kafka_producer = KafkaProducer(kafka_config)
+        return cls(kafka_producer, topic, key, drcb)
+    
+    def close_spider(self, spider):
+        """
+        Flushes the queue when the spider is closed.
+        """
+        logging.info("Flushing Kafka publish queue...")
+        self.producer.flush()
+        logging.info("Kafka publish queue flushed.")
+        self.producer.close()
diff --git a/kafka_scrapy_connect/spiders.py b/kafka_scrapy_connect/spiders.py
new file mode 100644
index 0000000..82eb016
--- /dev/null
+++ b/kafka_scrapy_connect/spiders.py
@@ -0,0 +1,182 @@
+# spiders.py
+from scrapy import signals
+from scrapy.http import Request
+from scrapy.exceptions import DontCloseSpider
+from scrapy.spiders import Spider
+from confluent_kafka import Consumer, KafkaException, Producer
+from scrapy.spidermiddlewares.httperror import HttpError
+from twisted.internet.error import DNSLookupError
+from twisted.internet.error import TimeoutError, TCPTimedOutError
+from urllib.parse import urlparse
+import os
+import sys
+import logging
+import json
+from datetime import datetime
+
+
+class KafkaSpiderMixin:
+    """
+    Mixin class to implement reading URLs from a Kafka queue.
+    """
+
+    def setup_kafka_producer(self, settings):
+        kafka_config = {'bootstrap.servers': settings.get('SCRAPY_KAFKA_HOSTS', 'localhost:9092')}
+        self.producer = Producer(kafka_config)
+        self.network_error_topic = settings.get('SCRAPY_ERROR_TOPIC', 'ScrapyNetworkErrors')
+
+    def process_kafka_message(self, message, meta={},headers={}):
+        """
+        Extracts URLs from a Kafka message.
+        :param message: A Kafka message object
+        :type message: Any
+        :return: URL extracted from the message, or None if extraction fails
+        :rtype: str or None
+        """
+        # Process message
+        try:
+            if message is None:
+                return None
+            if message.error():
+                # Handle Kafka errors
+                if message.error().code() == KafkaError._PARTITION_EOF:
+                    # End of partition
+                    return None
+                else:
+                    raise KafkaException(message.error())
+            else:
+                try:
+                    # Attempt to decode message as JSON
+                    json_obj = json.loads(message.value())
+                    # If decoding succeeds, it's a JSON object
+                    if isinstance(json_obj, dict) and 'url' in json_obj:
+                        url = json_obj['url']
+                        if self.is_valid_url(url):
+                            logging.debug(f"Received valid URL => {url}")
+                            return url, meta, headers
+                        else:
+                            logging.warning(f"Invalid URL => {url}")
+                            return None
+                    else:
+                        logging.warn(f"Invalid JSON format => {json_obj}")
+                        return None
+                except  json.decoder.JSONDecodeError:
+                    message = message.value().decode()
+                    # If decoding as JSON fails, check if it's a valid URL string
+                    if self.is_valid_url(message):
+                        logging.debug(f"Received valid URL => {message}")
+                        return message, meta, headers
+                    else:
+                        logging.warn(f"Message is not a valid URL => {message}")
+                        return None
+        except ValueError as e:
+            logging.warn(f"Error processing message: {e}")
+            return None
+
+    def is_valid_url(self, url):
+        """
+        Checks if a URL is valid.
+        :param url: URL to be validated
+        :type url: str
+        :return: True if the URL is valid, False otherwise
+        :rtype: bool
+        """
+        parsed_url = urlparse(url)
+        return parsed_url.scheme in ['http', 'https'] and bool(parsed_url.netloc)
+
+    def setup_kafka_consumer(self, settings):
+        """
+        Sets up the Kafka consumer.
+        :param settings: The current Scrapy settings being used
+        :type settings: scrapy.settings.Settings
+        """
+        kafka_hosts = settings.get('SCRAPY_KAFKA_HOSTS', 'localhost:9092')
+        consumer_config = settings.get('SCRAPY_CONSUMER_CONFIG', {})
+        kafka_config = {'bootstrap.servers': kafka_hosts, **consumer_config}
+        if 'group.id' not in kafka_config:
+            kafka_config['group.id'] = 'kafka-scrapy'
+        topic = settings.get('SCRAPY_INPUT_TOPIC', 'ScrapyInput')
+        try:
+            self.consumer = Consumer(kafka_config)
+            self.consumer.subscribe([topic])
+            logging.info(f'Instantiated a kafka consumer subscribed to topic: {topic} with the following configuration: {kafka_config}')
+        except KafkaException as e:
+            logging.error(f"Failed to connect to Kafka: {e}")
+            sys.exit(1)
+        # Call idle signal when there are no requests left
+        self.crawler.signals.connect(self.spider_idle, signal=signals.spider_idle)
+        self.crawler.signals.connect(self.item_scraped, signal=signals.item_scraped)
+        logging.info(f'Connected with the following config: {kafka_config}')
+        logging.info(f"Listening to Kafka topic {settings.get('SCRAPY_INPUT_TOPIC')} for incoming messages.")
+
+    def fetch_next_request(self):
+        """
+        Fetches the next request to be scheduled
+        Consumes messages from Kafka.
+        :rtype: scrapy.Request or None
+        """
+        batch_size = int(os.getenv('KAFKA_BATCH_SIZE') or '1')
+        messages = self.consumer.consume(num_messages=batch_size, timeout=1.0)
+        if not messages or len(messages) == 0:
+            logging.debug('No messages to process')
+            return None
+        logging.info(f"Received batch with {len(messages)} messages.")
+        for message in messages:
+            result = self.process_kafka_message(message)
+            if result is not None:
+            #if request_url:
+                url,meta,headers = result
+                logging.info(f'Crawling {url}')
+                yield Request(url=url, meta=meta,headers=headers,dont_filter=True, errback=self.network_error_cb)
+
+    def network_error_cb(self, failure):
+        error = f'Network request failure: {repr(failure)}'
+        url = failure.request.url
+        logging.warning(f'{error}')
+        payload = {'url': url, 'error': error}
+        self.publish_failures(self.network_error_topic, payload)
+
+    def publish_failures(self, topic, payload):
+        logging.info(f'Publishing failure to {topic}')
+        self.producer.produce(topic, value=json.dumps(payload))
+        self.producer.poll()
+
+    def schedule_next_request(self):
+        """Schedules the next request if available."""
+        for request in self.fetch_next_request():
+            self.crawler.engine.crawl(request)
+            logging.debug("Scheduled next request.")
+
+    def spider_idle(self):
+        """Schedules a request if available, otherwise waits."""
+        self.schedule_next_request()
+        logging.debug("Spider is idle.")
+        raise DontCloseSpider
+
+    def item_scraped(self, *args, **kwargs):
+        """Schedules the next request after an item has been scraped."""
+        self.schedule_next_request()
+
+    def close_spider(self, spider):
+        """
+        Flushes the queue when the spider is closed.
+        """
+        logging.info("Flushing Kafka publish queue...")
+        self.producer.flush()
+        logging.info("Kafka publish queue flushed.")
+        self.producer.close()
+
+
+class KafkaListeningSpider(KafkaSpiderMixin, Spider):
+    """
+    Spider that listens to a Kafka topic for incoming messages and initiates crawling.
+    """
+
+    def _set_crawler(self, crawler):
+        """
+        Sets up the crawler.
+        :type crawler: scrapy.crawler.Crawler
+        """
+        super(KafkaListeningSpider, self)._set_crawler(crawler)
+        self.setup_kafka_consumer(crawler.settings)
+        self.setup_kafka_producer(crawler.settings)
diff --git a/requirements.txt b/requirements.txt
new file mode 100644
index 0000000..8fbeb9f
--- /dev/null
+++ b/requirements.txt
@@ -0,0 +1,6 @@
+scrapy>=2.11.0
+confluent-kafka>=2.3.0
+pytest>=8.0.1
+pytest-mock>=3.12.0
+pytest-emoji>=0.2.0
+pytest-md-report>=0.5.1
\ No newline at end of file
diff --git a/setup.py b/setup.py
new file mode 100644
index 0000000..912ac47
--- /dev/null
+++ b/setup.py
@@ -0,0 +1,27 @@
+import os
+from setuptools import setup, find_packages
+
+MY_DIR = os.path.dirname(__file__)
+README_MD = open(os.path.join(MY_DIR, 'README.md')).read()
+
+setup(
+    name='kafka-scrapy-connect',
+    version='{{VERSION_PLACEHOLDER}}',
+    description='Integrating Scrapy with kafka using the confluent-kafka python client',
+    long_description=README_MD,
+    long_description_content_type="text/markdown",
+    packages=['kafka_scrapy_connect'],
+    install_requires=[
+        'scrapy >= 2.11.1',
+        'confluent-kafka >= 2.2.0'
+    ],
+    url='https://github.com/spicyparrot/kafka_scrapy_connect',
+    classifiers = [
+        "Intended Audience :: Developers",
+        "License :: OSI Approved :: MIT License",
+        "Programming Language :: Python :: 3.9",
+        "Programming Language :: Python :: 3.10",
+        "Programming Language :: Python :: 3.11"
+    ],
+    keywords="kafka, scrapy, crawling, scraping",
+)


Test patch:
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
new file mode 100644
index 0000000..f26eab9
--- /dev/null
+++ b/.github/workflows/test.yml
@@ -0,0 +1,58 @@
+name: 🧪 Testing
+
+concurrency:
+  group: ${{ github.ref_name }}
+  cancel-in-progress: true
+
+on:
+  push:
+    branches:
+      - "**" # matches every branch  (!trunk removed to ensure main branch SonarCloud analysis is performed)
+
+env:
+  ENVIRONMENT: "ci"
+  RED: '\033[0;31m'
+  GREEN: '\033[0;32m'
+  YELLOW: '\033[0;33m'
+  PURPLE: '\033[0;35m'
+  WHITE: '\033[0m'
+
+jobs:
+  test-repo:
+    name: 👩‍💻 Unit tests
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        python_version: ["3.9", "3.10", "3.11"]
+    steps:
+      - name: 🛀 Checkout Self
+        uses: actions/checkout@v4
+
+      - name: 🐍 Python Setup
+        uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python_version }}
+          cache: "pip" # caching pip dependencies
+
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r requirements.txt
+
+      - name: 🧠 Bring up CI Kafka cluster
+        uses: spicyparrot/kafka-kraft-action@v1.1.0
+        with:
+          kafka-version: "3.6.1"
+          kafka-topics: "ScrapyInput,1,ScrapyOutput,1,ScrapyError,1"
+
+      - name: 🧪 Run Kafka Scrapy tests
+        id: tests
+        run: |
+          pytest -s -v --emoji tests/test_kafka_scrapy_connect.py --disable-warnings --md-report --md-report-flavor gfm --md-report-output md_report.md
+        env:
+          KAFKA_BOOTSTRAP_SERVERS: ${{ env.kafka_runner_address }}:9093
+
+      - name: 📋 Adding markdown table
+        run: |
+          cat md_report.md >> $GITHUB_STEP_SUMMARY
diff --git a/.github/workflows/test_pypi.yml b/.github/workflows/test_pypi.yml
new file mode 100644
index 0000000..9cbe8a0
--- /dev/null
+++ b/.github/workflows/test_pypi.yml
@@ -0,0 +1,41 @@
+name: Publish Python 🐍 distributions 📦 to PyPI
+
+on:
+  # Triggers the workflow when a release is created
+  release:
+    types: [created]
+  # Allows you to run this workflow manually from the Actions tab
+  workflow_dispatch:
+
+jobs:
+  build-and-publish:
+    name: Build and publish Python 🐍 distributions 📦 to testPyPI
+    runs-on: ubuntu-18.04
+    steps:
+      - uses: actions/checkout@master
+      - name: Set up Python 3.11
+        uses: actions/setup-python@v4
+        with:
+          python-version: "3.11"
+
+      - name: Install pypa/setuptools
+        run: >-
+          python -m
+          pip install wheel
+
+      - name: Extract tag name
+        id: tag
+        run: echo ::set-output name=TAG_NAME::$(echo $GITHUB_REF | cut -d / -f 3)
+      - name: Update version in setup.py
+        run: >-
+          sed -i "s/{{VERSION_PLACEHOLDER}}/${{ steps.tag.outputs.TAG_NAME }}/g" setup.py
+
+      - name: Build a binary wheel
+        run: >-
+          python setup.py sdist bdist_wheel
+
+      - name: Publish distribution 📦 to TestPyPI
+        uses: pypa/gh-action-pypi-publish@release/v1
+        with:
+          password: ${{ secrets.TESTPYPI_TOKEN }}
+          repository-url: https://test.pypi.org/legacy/
diff --git a/tests/test_kafka_scrapy_connect.py b/tests/test_kafka_scrapy_connect.py
new file mode 100644
index 0000000..958806f
--- /dev/null
+++ b/tests/test_kafka_scrapy_connect.py
@@ -0,0 +1,164 @@
+from ast import parse
+import os
+import sys
+import json
+import uuid
+import unittest
+import pytest
+import requests
+import time
+from confluent_kafka import Producer, Consumer, KafkaError
+from scrapy.http import HtmlResponse, Response, Request
+from scrapy.settings import Settings
+
+SCRIPT_DIRECTORY = os.path.dirname(os.path.realpath(__file__))
+sys.path.insert(0, os.path.abspath(f'{SCRIPT_DIRECTORY}/../'))
+
+from kafka_scrapy_connect.spiders import KafkaListeningSpider
+#from kafka_scrapy_connect.spiders import KafkaSpiderMixin
+from kafka_scrapy_connect.pipelines import KafkaPublishPipeline
+
+class TestSpider(KafkaListeningSpider):
+    name = 'test-spider'
+
+    def process_kafka_input_message(self, message):
+        msg = json.loads(message.value())
+        request_url = msg['url']
+        return request_url
+    
+    def parse(self, response):
+        for quote in response.xpath('//div[@class="quote"]'):
+            yield {
+                'text' : quote.xpath('./span[@class="text"]/text()').extract_first(),
+                'author' : quote.xpath('.//small[@class="author"]/text()').extract_first(),
+                'tags' : quote.xpath('.//div[@class="tags"]/a[@class="tag"]/text()').extract()
+            }
+
+#=============================================================================================#
+#===================================== PYTEST FIXTURES =======================================#
+#=============================================================================================#
+@pytest.fixture
+def kafka_producer():
+    conf = {'bootstrap.servers': os.getenv('KAFKA_BOOTSTRAP_SERVERS','localhost:29092')}
+    producer = Producer(conf)
+    return producer
+
+@pytest.fixture
+def kafka_consumer():
+    group_id = str(uuid.uuid4())
+    kafka_config = {
+        'bootstrap.servers' : os.getenv('KAFKA_BOOTSTRAP_SERVERS','localhost:29092'),
+        'group.id' : 'pytest-' + group_id,
+        'auto.offset.reset' : 'earliest'
+    }
+    consumer = Consumer(kafka_config)
+    return consumer
+
+@pytest.fixture
+def kafka_input_message(kafka_consumer):
+    kafka_consumer.subscribe(['ScrapyInput'])
+    message = kafka_consumer.poll(timeout=30)
+    return message
+
+@pytest.fixture
+def kafka_output_message(kafka_consumer):
+    kafka_consumer.subscribe(['ScrapyOutput'])
+    message = kafka_consumer.poll(timeout=30)
+    message = json.loads(message.value())
+    return message
+
+@pytest.fixture
+def html_response_fixture():
+    url = 'https://quotes.toscrape.com'
+    response = requests.get(url)
+    response.raise_for_status()
+    test_directory = os.path.dirname(os.path.abspath(__file__))
+    fixture_path = os.path.join(test_directory, 'test_response.html')
+    with open(fixture_path, 'wb') as f:
+        f.write(response.content)
+    yield fixture_path
+    os.remove(fixture_path)
+
+@pytest.fixture
+def parse_result(html_response_fixture):
+    spider = TestSpider()
+    with open(html_response_fixture, 'rb') as f:
+        html_content = f.read()
+    mock_response = create_mock_response(html_content)
+    result = list(spider.parse(mock_response))
+    yield result
+
+def create_mock_response(html_content, url='https://quotes.toscrape.com',status=200, meta={}):
+    meta_dict = meta
+    request = Request(url=url, meta=meta_dict)
+    return HtmlResponse(url, status=status, request=request, body=html_content,encoding='utf-8')
+
+#=============================================================================================#
+#====================================== PYTEST TESTS =========================================#
+#=============================================================================================#
+
+def test_publish_valid_message_to_kafka(kafka_producer):
+    '''
+    Verifies if a message can be successfully published to a Kafka topic. 
+    It checks if the message is successfully delivered by setting a flag 
+    upon delivery and then checking if the flag is set to true.
+    '''
+    message_delivered = False
+    message_payload = {"url" : "https://quotes.toscrape.com"}
+    message_json = json.dumps(message_payload)
+    def delivery_callback(err, msg):
+        nonlocal message_delivered
+        if err is not None:
+            raise Exception(f"Failed to deliver message: {err}")
+        else:
+            message_delivered = True
+    kafka_producer.produce('ScrapyInput', value=message_json.encode('utf-8'),callback=delivery_callback)
+    kafka_producer.flush()
+    assert message_delivered, "Message was not delivered successfully"
+
+def test_consume_valid_message_from_kafka(kafka_input_message):
+    '''
+    This test checks if a message can be consumed from a Kafka topic. 
+    It simply checks if a message object is received from the Kafka topic.
+    '''
+    assert kafka_input_message is not None, "No message received from Kafka topic"
+
+def test_is_valid_url():
+    spider = TestSpider()
+    assert spider.is_valid_url('https://quotes.toscrape.com')
+
+def test_is_invalid_url():
+    spider = TestSpider()
+    assert spider.is_valid_url('im_not_a_url') == False
+
+def test_process_kafka_input_message(kafka_input_message):
+    '''
+     This test checks if the method process_kafka_input_message of the TestSpider class 
+     correctly processes a Kafka message by extracting the URL from the message
+    '''
+    spider = TestSpider()
+    result = spider.process_kafka_input_message(kafka_input_message)
+    assert result == 'https://quotes.toscrape.com'
+
+def test_parse(parse_result):
+    '''
+    This test verifies if the parse method of the TestSpider class correctly extracts data from an HTML response.
+    '''
+    result = parse_result
+    assert result[0]['author'] == 'Albert Einstein'
+    assert result[0]['tags'] == ['change','deep-thoughts','thinking','world']
+    assert len(result[0]['text']) > 1
+
+def test_pipeline_publish_to_kafka(kafka_producer,parse_result,kafka_consumer):
+    '''
+    This test checks if the KafkaPublishPipeline's process_item method executes without errors.
+    '''
+    kafka_publish_pipeline = KafkaPublishPipeline(producer=kafka_producer, topic='ScrapyOutput')
+    spider = TestSpider()
+    kafka_publish_pipeline.process_item(parse_result[0],spider)
+    kafka_producer.flush()
+    kafka_consumer.subscribe(['ScrapyOutput'])
+    message = kafka_consumer.poll(timeout=30)
+    message = json.loads(message.value())
+    assert isinstance(message, dict)
+    assert message is not None, "No output message received"
\ No newline at end of file