kafka init (#13)

.github/workflows/build_and_publish.yml

@@ -58,7 +58,8 @@ jobs:
       name: Testing
       url: https://api.test.profcomff.com/userdata
     env:
-      CONTAINER_NAME: com_profcomff_api_userdata_test
+      API_CONTAINER_NAME: com_profcomff_api_userdata_test
+      WORKER_CONTAINER_NAME: com_profcomff_worker_userdata_test
     permissions:
       packages: read
 
@@ -72,25 +73,43 @@ jobs:
             --rm \
             --network=web \
             --env DB_DSN=${{ secrets.DB_DSN }} \
-            --name ${{ env.CONTAINER_NAME }}_migration \
+            --name ${{ env.API_CONTAINER_NAME }}_migration \
             --workdir="/" \
             ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:test \
             alembic upgrade head
 
-      - name: Run new version
-        id: run_prod
+      - name: Run new version API
+        id: run_test_api
         run: |
-          docker stop ${{ env.CONTAINER_NAME }} || true && docker rm ${{ env.CONTAINER_NAME }} || true
+          docker stop ${{ env.API_CONTAINER_NAME }} || true && docker rm ${{ env.API_CONTAINER_NAME }} || true
           docker run \
             --detach \
             --restart always \
             --network=web \
             --env DB_DSN='${{ secrets.DB_DSN }}' \
             --env ROOT_PATH='/userdata' \
             --env GUNICORN_CMD_ARGS='--log-config logging_test.conf' \
-            --name ${{ env.CONTAINER_NAME }} \
+            --name ${{ env.API_CONTAINER_NAME }} \
             ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:test
 
+      - name: Run new version worker
+        id: run_test_worker
+        run: |
+          docker stop ${{ env.WORKER_CONTAINER_NAME }} || true && docker rm ${{ env.WORKER_CONTAINER_NAME }} || true
+          docker run \
+            --detach \
+            --restart always \
+            --network=web \
+            --env DB_DSN='${{ secrets.DB_DSN }}' \
+            --env KAFKA_DSN='${{ secrets.KAFKA_DSN }}' \
+            --env KAFKA_LOGIN='${{ secrets.KAFKA_LOGIN }}' \
+            --env KAFKA_PASSWORD='${{ secrets.KAFKA_PASSWORD }}' \
+            --env KAFKA_GROUP_ID='${{ vars.KAFKA_GROUP_ID }}' \
+            --env KAFKA_TOPICS='${{ vars.KAFKA_TOPICS }}' \
+            --name ${{ env.WORKER_CONTAINER_NAME }} \
+            ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:test python -m userdata_api start --instance worker
+      
+
   deploy-production:
     name: Deploy Production
     needs: build-and-push-image
@@ -100,7 +119,8 @@ jobs:
       name: Production
       url: https://api.profcomff.com/userdata
     env:
-      CONTAINER_NAME: com_profcomff_api_userdata
+      API_CONTAINER_NAME: com_profcomff_api_userdata
+      WORKER_CONTAINER_NAME: com_profcomff_worker_userdata
     permissions:
       packages: read
 
@@ -114,21 +134,38 @@ jobs:
             --rm \
             --network=web \
             --env DB_DSN=${{ secrets.DB_DSN }} \
-            --name ${{ env.CONTAINER_NAME }}_migration \
+            --name ${{ env.API_CONTAINER_NAME }}_migration \
             --workdir="/" \
             ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:latest \
             alembic upgrade head
 
-      - name: Run new version
-        id: run_test
+      - name: Run new version API
+        id: run_prod_api
         run: |
-          docker stop ${{ env.CONTAINER_NAME }} || true && docker rm ${{ env.CONTAINER_NAME }} || true
+          docker stop ${{ env.API_CONTAINER_NAME }} || true && docker rm ${{ env.API_CONTAINER_NAME }} || true
           docker run \
             --detach \
             --restart always \
             --network=web \
             --env DB_DSN='${{ secrets.DB_DSN }}' \
             --env ROOT_PATH='/userdata' \
             --env GUNICORN_CMD_ARGS='--log-config  logging_prod.conf' \
-            --name ${{ env.CONTAINER_NAME }} \
+            --name ${{ env.API_CONTAINER_NAME }} \
             ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:latest
+
+      - name: Run new version worker
+        id: run_prod_worker
+        run: |
+          docker stop ${{ env.WORKER_CONTAINER_NAME }} || true && docker rm ${{ env.WORKER_CONTAINER_NAME }} || true
+          docker run \
+            --detach \
+            --restart always \
+            --network=web \
+            --env DB_DSN='${{ secrets.DB_DSN }}' \
+            --env KAFKA_DSN='${{ secrets.KAFKA_DSN }}' \
+            --env KAFKA_LOGIN='${{ secrets.KAFKA_LOGIN }}' \
+            --env KAFKA_PASSWORD='${{ secrets.KAFKA_PASSWORD }}' \
+            --env KAFKA_GROUP_ID='${{ vars.KAFKA_GROUP_ID }}' \
+            --env KAFKA_TOPICS='${{ vars.KAFKA_TOPICS }}' \
+            --name ${{ env.WORKER_CONTAINER_NAME }} \
+            ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:latest python -m userdata_api start --instance worker

README.md

@@ -21,8 +21,19 @@
     ```
 4. Запускайте приложение!
     ```console
-    foo@bar:~$ python -m userdata_api
+    foo@bar:~$ python -m userdata_api start --instance api -- запустит АПИ
+    foo@bar:~$ python -m userdata_api start --instance worker -- запустит Kafka worker
     ```
+   
+Приложение состоит из двух частей - АПИ и Kafka worker'а. 
+
+АПИ нужно для управления структурой пользовательских данных - 
+контроль над категориями данных, параметрами, источниками данных.
+Также, в АПИ пользовательские данные может слать
+сам пользователь(владелец этих данных), а также админ
+
+Kafka worker нужен для того, чтобы разгребать поступающие от OAuth 
+методов авторизации AuthAPI пользовательские данные
 
 ## ENV-file description
 - `DB_DSN=postgresql://postgres@localhost:5432/postgres` – Данные для подключения к БД

migrations/env.py

@@ -3,8 +3,8 @@
 from alembic import context
 from sqlalchemy import engine_from_config, pool
 
+from settings import get_settings
 from userdata_api.models.base import Base
-from userdata_api.settings import get_settings
 
 
 # this is the Alembic Config object, which provides

requirements.txt

@@ -9,3 +9,5 @@ pydantic[dotenv]
 SQLAlchemy
 uvicorn
 pydantic-settings
+event_schema_profcomff
+confluent_kafka

userdata_api/settings.py → settings.py

@@ -9,6 +9,13 @@ class Settings(BaseSettings):
     """Application settings"""
 
     DB_DSN: PostgresDsn = 'postgresql://postgres@localhost:5432/postgres'
+
+    KAFKA_DSN: str | None = None
+    KAFKA_LOGIN: str | None = None
+    KAFKA_PASSWORD: str | None = None
+    KAFKA_TOPICS: list[str] | None = None
+    KAFKA_GROUP_ID: str | None = None
+
     ROOT_PATH: str = '/' + os.getenv("APP_NAME", "")
 
     CORS_ALLOW_ORIGINS: list[str] = ['*']

tests/conftest.py

@@ -3,9 +3,9 @@
 from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker
 
+from settings import get_settings
 from userdata_api.models.db import *
 from userdata_api.routes.base import app
-from userdata_api.settings import get_settings
 from userdata_api.utils.utils import random_string
 
 

tests/test_worker/test_worker.py

@@ -0,0 +1,110 @@
+import pytest
+import sqlalchemy.exc
+from event_schema.auth import UserLogin
+
+from userdata_api.models.db import Category, Info, Param, Source
+from userdata_api.utils.utils import random_string
+from worker.user import patch_user_info
+
+
+@pytest.fixture()
+def category(dbsession):
+    name = f"test{random_string()}"
+    dbsession.add(
+        _cat := Category(
+            name=name, read_scope=f"testscope.{random_string()}", update_scope=f"testscope.{random_string()}"
+        )
+    )
+    dbsession.commit()
+    yield _cat
+    dbsession.delete(_cat)
+    dbsession.commit()
+
+
+@pytest.fixture()
+def param(dbsession, category):
+    time_ = f"test{random_string()}"
+    dbsession.add(
+        _par := Param(name=f"test{time_}", category_id=category.id, type="last", changeable=True, is_required=True)
+    )
+    dbsession.commit()
+    yield _par
+    dbsession.delete(_par)
+    dbsession.commit()
+
+
+@pytest.fixture()
+def source(dbsession):
+    time_ = f"test{random_string()}"
+    __source = Source(name=f"test{time_}", trust_level=8)
+    dbsession.add(__source)
+    dbsession.commit()
+    yield __source
+    dbsession.delete(__source)
+    dbsession.commit()
+
+
+@pytest.fixture()
+def info(param, source, dbsession):
+    time_ = f"test{random_string()}"
+    __info = Info(value=f"test{time_}", source_id=source.id, param_id=param.id, owner_id=1)
+    dbsession.add(__info)
+    dbsession.commit()
+    yield __info
+    try:
+        dbsession.delete(__info)
+        dbsession.commit()
+    except sqlalchemy.exc.Any:
+        pass
+
+
+def test_create(param, source, dbsession):
+    with pytest.raises(sqlalchemy.exc.NoResultFound):
+        dbsession.query(Info).filter(Info.param_id == param.id, Info.source_id == source.id, Info.value == "test").one()
+    patch_user_info(
+        UserLogin.model_validate(
+            {"items": [{"category": param.category.name, "param": param.name, "value": "test"}], "source": source.name}
+        ),
+        1,
+        session=dbsession,
+    )
+    info = (
+        dbsession.query(Info).filter(Info.param_id == param.id, Info.source_id == source.id, Info.value == "test").one()
+    )
+    assert info
+    dbsession.delete(info)
+    dbsession.commit()
+
+
+def test_update(info, dbsession):
+    assert info.value != "updated"
+    patch_user_info(
+        UserLogin.model_validate(
+            {
+                "items": [{"category": info.category.name, "param": info.param.name, "value": "updated"}],
+                "source": info.source.name,
+            }
+        ),
+        1,
+        session=dbsession,
+    )
+
+    dbsession.expire(info)
+    assert info.value == "updated"
+
+
+def test_delete(info, dbsession):
+    assert info.is_deleted is False
+    patch_user_info(
+        UserLogin.model_validate(
+            {
+                "items": [{"category": info.category.name, "param": info.param.name, "value": None}],
+                "source": info.source.name,
+            }
+        ),
+        1,
+        session=dbsession,
+    )
+
+    dbsession.expire(info)
+    assert info.is_deleted is True

userdata_api/__main__.py

@@ -1,7 +1,25 @@
+import argparse
+
 import uvicorn
 
 from userdata_api.routes.base import app
+from worker.consumer import process
+
+
+def get_args():
+    parser = argparse.ArgumentParser()
+    subparsers = parser.add_subparsers(dest='command')
+
+    start = subparsers.add_parser("start")
+    start.add_argument('--instance', type=str, required=True)
+
+    return parser.parse_args()
 
 
 if __name__ == '__main__':
-    uvicorn.run(app)
+    args = get_args()
+    match args.instance:
+        case "api":
+            uvicorn.run(app)
+        case "worker":
+            process()

userdata_api/routes/base.py

@@ -2,8 +2,8 @@
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi_sqlalchemy import DBSessionMiddleware
 
+from settings import get_settings
 from userdata_api import __version__
-from userdata_api.settings import get_settings
 
 from .category import category
 from .param import param

worker/consumer.py

@@ -0,0 +1,40 @@
+import logging
+from typing import Any
+
+import pydantic
+from event_schema.auth import UserLogin, UserLoginKey
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from settings import get_settings
+from worker.kafka import KafkaConsumer
+
+from .user import patch_user_info
+
+
+log = logging.getLogger(__name__)
+settings = get_settings()
+consumer = KafkaConsumer()
+
+_engine = create_engine(str(settings.DB_DSN), pool_pre_ping=True, isolation_level="AUTOCOMMIT")
+_Session = sessionmaker(bind=_engine)
+
+
+def process_models(key: Any, value: Any) -> tuple[UserLoginKey | None, UserLogin | None]:
+    try:
+        return UserLoginKey.model_validate(key), UserLogin.model_validate(value)
+    except pydantic.ValidationError:
+        log.error(f"Validation error occurred, {key=}, {value=}", exc_info=False)
+        return None, None
+
+
+def process_message(message: tuple[Any, Any]) -> None:
+    processed_k, processed_v = process_models(*message)
+    if not (processed_k and processed_v):
+        return
+    patch_user_info(processed_v, processed_k.user_id, session=_Session())
+
+
+def process():
+    for message in consumer.listen():
+        process_message(message)