INIT

Author: pjabadesco

.env

@@ -0,0 +1,9 @@
+DATABASE_URL=postgresql+psycopg2://postgres:password@db:5432/post_db
+DB_USER=postgres
+DB_PASSWORD=password
+DB_NAME=post_db 
+PGADMIN_EMAIL=[email protected]
+PGADMIN_PASSWORD=admin
+SECRET_KEY=89023f0j2039jf90jkakmciumcses9mcs99mcs93c48ym72tmarf737mc7h47wna4c8791289h8f1h3f8h376fh8bf4
+CELERY_BROKER_URL=redis://redis:6379/0
+CELERY_RESULT_BACKEND=redis://redis:6379/0

Dockerfile

@@ -0,0 +1,8 @@
+FROM python:3.9.6-slim-buster
+ENV PYTHONUNBUFFERED=1
+WORKDIR /app
+COPY requirements.txt requirements.txt
+RUN pip3 install --upgrade pip
+RUN pip3 install -r requirements.txt
+COPY . /app
+EXPOSE 8000

README.md

@@ -0,0 +1,26 @@
+https://www.youtube.com/watch?v=mcX_4EvYka4
+https://github.com/veryacademy/YT_FastAPI_Celery_Redis_Flower_Introduction
+
+docker-compose build
+docker-compose up
+
+conda env list
+conda create --name taskqueue-http python=3.9
+conda activate taskqueue-http
+conda env remove --name taskqueue-http
+pip install -r requirements.txt
+
+pip3 install fastapi
+pip3 install uvicorn
+pip3 install orjson
+pip3 freeze > requirements.txt
+python3 ./app.py
+uvicorn main:app --host 0.0.0.0 --port 8001
+
+curl -X POST http://localhost:8000/ -H "Content-Type: application/json" --data '{"name":"test", "url": "https://bahay.ph", "http_method": "GET"}'
+
+curl -X POST http://localhost:8000/ -H "Content-Type: application/json" --data '{"name":"test", "url": "https://bahay.ph", "http_method": "GET"}'
+
+curl -X POST http://localhost:8000/ -H "Content-Type: application/json" --data '{"name":"test", "url": "https://api.neuracash.com/transactions", "http_method": "GET"}'
+
+

__pycache__/celery_worker.cpython-39.pyc

@@ -0,0 +1,55 @@
+import os
+import time
+import requests
+from requests.exceptions import RequestException
+import json
+
+
+from celery import Celery
+from dotenv import load_dotenv
+
+load_dotenv(".env")
+
+celery = Celery(__name__)
+celery.conf.broker_url = os.environ.get("CELERY_BROKER_URL")
+celery.conf.result_backend = os.environ.get("CELERY_RESULT_BACKEND")
+
+@celery.task(name="create_task", bind=True, autoretry_for=(RequestException,), retry_backoff=True)
+def create_task(self, url, http_method, body, headers):
+    # try:
+        if http_method == "GET":
+            response = requests.get(url, headers=headers)
+        elif http_method == "POST":
+            response = requests.post(url, data=body, headers=headers)
+        elif http_method == "PUT":
+            response = requests.put(url, data=body, headers=headers)
+        elif http_method == "PATCH":
+            response = requests.patch(url, data=body, headers=headers)
+        elif http_method == "DELETE":
+            response = requests.delete(url, headers=headers)
+        else:
+            raise RequestException("HTTP Method not supported")
+
+        # GET response http code
+        # response_headers = {}
+        # for key, value in response.headers.items():
+        #     response_headers[key] = value
+
+        if not response.ok:
+            raise RequestException(f'{url} returned unexpected response code: {response.status_code}')
+
+        return {
+            "url": url, 
+            "http_method": http_method, 
+            "headers": headers, 
+            "body": body, 
+            "response_code": response.status_code, 
+            # "response_headers": response_headers, 
+            "response_body": response.text
+        }
+        # return response.json()
+        # return str(response)
+        
+    # except Exception as e:
+    #     self.retry(countdown=5, exc=e)
+        # raise Exception(str(e))

__pycache__/main.cpython-39.pyc

@@ -0,0 +1,78 @@
+version: "3.8"
+
+services:
+
+  app:
+    container_name: app
+    build: .
+    command: bash -c "uvicorn main:app --host 0.0.0.0 --port 8000 --reload"
+    volumes:
+      - .:/app
+    ports:
+      - 8000:8000
+    restart: always
+
+  redis:
+    container_name: redis
+    image: redis:6.2-alpine
+
+  celery_worker_01:
+    container_name: celery_worker_01
+    build: .
+    command: celery -A celery_worker.celery worker --loglevel=info
+    volumes:
+      - .:/app
+    environment:
+      - CELERY_BROKER_URL=${CELERY_BROKER_URL}
+      - CELERY_RESULT_BACKEND=${CELERY_RESULT_BACKEND}
+    depends_on:
+      - app
+      - redis
+
+  # celery_worker_02:
+  #   container_name: celery_worker_02
+  #   build: .
+  #   command: celery -A celery_worker.celery worker --loglevel=info
+  #   volumes:
+  #     - .:/app
+  #   environment:
+  #     - CELERY_BROKER_URL=${CELERY_BROKER_URL}
+  #     - CELERY_RESULT_BACKEND=${CELERY_RESULT_BACKEND}
+  #   depends_on:
+  #     - app
+  #     - redis
+
+  # celery_worker_03:
+  #   container_name: celery_worker_03
+  #   build: .
+  #   command: celery -A celery_worker.celery worker --loglevel=info
+  #   volumes:
+  #     - .:/app
+  #   environment:
+  #     - CELERY_BROKER_URL=${CELERY_BROKER_URL}
+  #     - CELERY_RESULT_BACKEND=${CELERY_RESULT_BACKEND}
+  #   depends_on:
+  #     - app
+  #     - redis
+
+  flower:
+    container_name: flower
+    build: .
+    command: celery -A celery_worker.celery flower --port=5555
+    ports:
+      - 5556:5555
+    environment:
+      - CELERY_BROKER_URL=${CELERY_BROKER_URL}
+      - CELERY_RESULT_BACKEND=${CELERY_RESULT_BACKEND}
+    depends_on:
+      - app
+      - redis
+      - celery_worker_01
+      # - celery_worker_02
+      # - celery_worker_03
+
+  redisinsight:
+    container_name: redisinsight
+    image: redislabs/redisinsight:latest
+    ports:
+      - 8001:8001

celery_worker.py

@@ -0,0 +1,72 @@
+from fastapi import Body, FastAPI
+from fastapi.responses import JSONResponse
+import uvicorn
+import json
+
+from celery_worker import create_task
+
+
+app = FastAPI()
+
+@app.get("/")
+def index():
+    return JSONResponse({"message": "Hello World"})
+
+@app.post("/")
+def index_post(data=Body(...)):
+    http_method = data.get("http_method", "GET")
+    url = data.get("url", "")
+    body = data.get("body", "{}")
+    headers = data.get("headers", '{"Content-Type": "application/json"}')
+    expires = data.get("expires", 86400)
+    name = data.get("name", "celery_worker_queue")
+
+    if len(url) == 0:
+        return JSONResponse({"message": "URL is required"}, status_code=500)
+    if not url.startswith("http://") and not url.startswith("https://"):
+        return JSONResponse({"message": "URL must start with http:// or https://"}, status_code=500)
+
+    if http_method not in ["GET", "POST", "PUT", "PATCH", "DELETE"]:
+        return JSONResponse({"message": "HTTP Method must be GET, POST, PUT, PATCH, or DELETE"}, status_code=500)
+
+    if len(body) > 0:
+        try:
+            json.loads(body)
+        except:
+            return JSONResponse({"message": "Body must be in JSON format"}, status_code=500)
+
+    if len(headers) > 0:
+        try:
+            headers = json.loads(headers)
+        except:
+            return JSONResponse({"message": "Headers must be in JSON format"}, status_code=500)
+
+    try:
+        expires = int(expires)
+    except:
+        return JSONResponse({"message": "Dispatch deadline must be an integer"}, status_code=500)
+    if expires < 0:
+        return JSONResponse({"message": "Dispatch deadline must be a positive integer"}, status_code=500)
+
+    try:
+        # create task
+        task = create_task.apply_async(
+            shadow_name=name,
+            args=(url, http_method, body, headers),
+            # max_retries=5,
+            # expires=expires
+        )
+        # task = create_task.delay(url, http_method, body, headers)
+    except Exception as e:
+        return JSONResponse({"message": str(e)}, status_code=500)
+
+    return JSONResponse(
+        {"message": "Task created successfully", "task_id": task.id}
+    )
+
+    # task = create_task.delay(http_method, url, body, headers, expires)
+    # # return JSONResponse({"Result": task.get()})
+    # return JSONResponse({"Result": task.ready()})
+
+# if __name__ == "__main__":
+#     uvicorn.run(app)

docker-compose.yml

@@ -0,0 +1,44 @@
+amqp==5.0.6
+anyio==3.4.0
+asgiref==3.4.1
+attrs==20.3.0
+billiard==3.6.4.0
+celery==5.2.1
+certifi==2020.12.5
+chardet==4.0.0
+click==8.0.3
+click-didyoumean==0.3.0
+click-plugins==1.1.1
+click-repl==0.2.0
+Deprecated==1.2.13
+fastapi==0.70.0
+flower==1.0.0
+h11==0.12.0
+humanize==3.12.0
+idna==2.10
+JsonForm==0.0.2
+jsonschema==3.2.0
+JsonSir==0.0.2
+kombu==5.2.2
+orjson==3.6.4
+prometheus-client==0.12.0
+prompt-toolkit==3.0.23
+pydantic==1.8.2
+pyrsistent==0.17.3
+python-dotenv==0.19.2
+Python-EasyConfig==0.1.7
+pytz==2021.3
+PyYAML==5.4.1
+redis==4.0.2
+requests==2.25.1
+Resource==0.2.1
+six==1.15.0
+sniffio==1.2.0
+starlette==0.16.0
+tornado==6.1
+typing_extensions==4.0.0
+urllib3==1.26.4
+uvicorn==0.15.0
+vine==5.0.0
+wcwidth==0.2.5
+wrapt==1.13.3