Celery tasks have been created

The data is fetched from ipdata.co and saved to the database by using background workers

Author: s3m3dov

api/__init__.py

@@ -1,11 +1,12 @@
 from fastapi import APIRouter
 
-from .endpoints import user, tasks, basic, users, items
+from .endpoints import user, tasks
+from .endpoints import tutorial, users, items
 
 api_router = APIRouter()
 api_router.include_router(user.router, tags=["user"])
 api_router.include_router(tasks.router, tags=["tasks"])
 
-api_router.include_router(basic.router, prefix="/basic", tags=["basic"])
+api_router.include_router(tutorial.router, prefix="/tutorial", tags=["tutorial"])
 api_router.include_router(users.router, prefix="/users", tags=["users"])
 api_router.include_router(items.router, prefix="/items", tags=["items"])

api/endpoints/basic.py

@@ -23,4 +23,4 @@ def user_items(Authorize: AuthJWT = Depends(), skip: int = 0, limit: int = 100):
     dependencies=[Depends(get_db)],
 )
 def create_item_for_user(user_id: int, item: ItemCreate):
-    return crud_items.create_user_item(item=item, user_id=user_id)
+    return crud_items.create_user_item(item=item, user_id=user_id)

api/endpoints/items.py

@@ -1,19 +1,28 @@
-from fastapi import APIRouter
-from pydantic import BaseModel
+from fastapi import APIRouter, Depends, Request
+from fastapi_jwt_auth import AuthJWT
 
-from celery_worker import add
+from celery.result import AsyncResult
+from celery_worker import celery as celery_app, user_ip_data_save
 
+from crud import ipdata as crud_ipdata
+from schemas.ipdata import IPDataCreate
 
 router = APIRouter()
 
 
-# Use pydantic to keep track of the input request payload
-class Numbers(BaseModel):
-    x: float
-    y: float
+@router.post('/task/')
+def enqueue_user_ip_data(ipdata: IPDataCreate, Authorize: AuthJWT = Depends()):
+    Authorize.jwt_required()
 
+    db_ipdata = crud_ipdata.get_by_ip_address(ipdata.ip_address)
+    if db_ipdata:
+        return {"msg": "The details of this IP exist in the database."}
 
-@router.post('/task/')
-def enqueue_add(n: Numbers):
-    # We use celery delay method in order to enqueue the task with the given parameters
-    add.delay(n.x, n.y)
+    task = user_ip_data_save.delay(ipdata.ip_address)
+    return {"msg": "Task has been created.", "task_id": task.id}
+
+
+@router.get("/status/{id}/")
+async def check_task_status(task_id: str):
+    result = AsyncResult(task_id, app=celery_app)
+    return {"status": result.state}

api/endpoints/tasks.py

@@ -0,0 +1,29 @@
+from fastapi import APIRouter, Request
+
+from celery_worker import add
+from schemas.tutorial import Numbers
+
+router = APIRouter()
+
+
+@router.get("/")
+async def root():
+    return {"message": "Hello World!"}
+
+
+@router.get("/hello/{name}")
+async def say_hello(name: str):
+    return {"message": f"Hello {name}"}
+
+
+@router.get("/my-ip/")
+async def show_user_ip(request: Request):
+    client_host = request.client.host
+    return {"client_host": client_host}
+
+
+@router.post('/sum/')
+def enqueue_add(n: Numbers):
+    add.delay(n.x, n.y)
+    # We use celery delay method in order to enqueue the
+    # task with the given parameters

api/endpoints/tutorial.py

@@ -1,7 +1,9 @@
 from celery import Celery
 from celery.utils.log import get_task_logger
-from core import settings
 
+from core import settings
+from db.models import IPData as model_IPData
+from crud.ipdata import get_user_ip_data
 
 celery = Celery(__name__)
 celery.conf.broker_url = settings.CELERY_BROKER_URL
@@ -10,6 +12,17 @@
 logger = get_task_logger(__name__)
 
 
+@celery.task
+def user_ip_data_save(ip_address):
+    ip_data_obj = model_IPData(
+        ip_address=ip_address,
+        ip_details=get_user_ip_data(ip_address),
+    )
+    ip_data_obj.save()
+
+    return ip_data_obj
+
+
 @celery.task
 def add(x, y):
     res = x + y

celery_worker.py

@@ -8,8 +8,8 @@
 
 class FastAPISettings:
     API_V1_STR: str = "/api/v1"
-
     IPDATA_API_KEY: str = env("IPDATA_API_KEY", default="test")
+
     AUTH_JWT_SECRET_KEY: str = env("AUTH_JWT_SECRET_KEY", default="secret")
     AUTH_JWT_TOKEN_ALGORITHM = "HS256"
     ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 1

core/config.py

@@ -1,13 +1,11 @@
 import json
 
 
-def json_prettify(json_dict):
-    return json.loads(
-        json.dumps(
-            json_dict,
-            ensure_ascii=False,
-            allow_nan=False,
-            indent=4,
-            separators=(", ", ": "),
-        ).encode("utf-8")
+def prettify_json(json_dict) -> str:
+    return json.dumps(
+        json_dict,
+        ensure_ascii=False,
+        allow_nan=False,
+        indent=4,
+        separators=(", ", ": "),
     )

core/utils.py

@@ -0,0 +1,28 @@
+import aiohttp
+import requests
+
+from core import settings
+from core.utils import prettify_json
+from db.models import IPData as model_IPData
+
+
+def get_by_ip_address(ip_address: str):
+    return model_IPData.filter(model_IPData.ip_address == ip_address).first()
+
+
+def get_user_ip_data(client_ip):
+    ip_data_url = f"https://api.ipdata.co/{client_ip}?api-key={settings.IPDATA_API_KEY}"
+    response = requests.get(ip_data_url)
+    return prettify_json(response.json())
+
+
+async def get_user_ip_data_async(client_ip):
+    ip_data_url = f"https://api.ipdata.co/{client_ip}?api-key={settings.IPDATA_API_KEY}"
+    async with aiohttp.ClientSession() as session:
+        async with session.get(ip_data_url) as response:
+            json_dict = await response.json()
+
+    return prettify_json(json_dict)
+
+
+

crud/ipdata.py

@@ -13,10 +13,18 @@ class Meta:
         database = mysql_db
 
 
+class IPData(peewee.Model):
+    ip_address = peewee.CharField()
+    ip_details = peewee.TextField()
+
+    class Meta:
+        database = mysql_db
+
+
 class Item(peewee.Model):
     title = peewee.CharField(index=True)
     description = peewee.CharField(index=True)
-    owner = peewee.ForeignKeyField(User, backref="items")
+    user = peewee.ForeignKeyField(User, backref="items")
 
     class Meta:
         database = mysql_db

db/models.py

@@ -11,7 +11,7 @@
 from api import api_router
 
 mysql_db.connect()
-mysql_db.create_tables([models.User, models.Item])
+mysql_db.create_tables([models.User, models.IPData, models.Item])
 mysql_db.close()
 
 app = FastAPI(

main.py

@@ -7,6 +7,7 @@ pymysql~=1.0.2
 cryptography~=36.0.1
 environs~=9.5.0
 aiohttp~=3.8.1
+requests~=2.27.1
 fastapi-jwt-auth~=0.5.0
 passlib[bcrypt]~=1.7.4
 celery~=5.2.3

requirements.txt

@@ -0,0 +1,22 @@
+from typing import Optional
+from pydantic import BaseModel
+
+from .utils import PeeweeGetterDict
+
+
+class IPDataBase(BaseModel):
+    ip_address: str
+
+
+class IPDataCreate(IPDataBase):
+    pass
+
+
+class IPData(IPDataCreate):
+    id: Optional[int] = None
+    ip_details: Optional[str]
+    # user_id: int
+
+    class Config:
+        orm_mode = True
+        getter_dict = PeeweeGetterDict

schemas/ipdata.py

@@ -15,7 +15,7 @@ class ItemCreate(ItemBase):
 
 class Item(ItemBase):
     id: int
-    owner_id: int
+    user_id: int
 
     class Config:
         orm_mode = True

schemas/item.py

@@ -0,0 +1,6 @@
+from pydantic import BaseModel
+
+
+class Numbers(BaseModel):
+    x: float
+    y: float

schemas/tutorial.py

@@ -3,6 +3,7 @@
 
 from .utils import PeeweeGetterDict
 from .item import Item
+from .ipdata import IPData
 
 
 class UserBase(BaseModel):
@@ -26,6 +27,7 @@ class UserCreate(UserBase):
 class UserInDBBase(UserBase):
     id: Optional[int] = None
     items: List[Item] = []
+    ipdata: IPData = None
 
     class Config:
         orm_mode = True