Пример создания универсального коннектора к источнику данных

FROM python:3.12.5-slim-bookworm
 
ARG UID=1000
ARG GID=0
 
# Создадим и настроим учётную запись пользователя
RUN useradd -m -u ${UID} -g ${GID} -s /bin/bash fmp
 
# Скопируем и установим Python-пакеты через poetry
RUN python -m pip install fastapi==0.112.2 uvicorn==0.30.6 python-multipart==0.0.20
COPY --chown=${UID}:${GID} src/ /home/fmp/
USER ${UID}
WORKDIR /home/fmp
CMD python main.py

from fastapi import FastAPI, UploadFile, File
from fastapi.security import HTTPBasic
import uvicorn
import csv
from io import StringIO
from typing import Dict
 
app = FastAPI()
security = HTTPBasic()
 
def clean_key(key: str) -> str:
    """Удаляет пробелы из ключа"""
    return key.strip().replace(" ", "")
 
def clean_value(value: str) -> str:
    """Удаляет пробелы из значения"""
    return value.strip() if isinstance(value, str) else value
 
def clean_row(row: Dict) -> Dict:
    """Очищает и ключи, и значения в строке"""
    return {clean_key(key): clean_value(value) for key, value in row.items()}
 
@app.post("/get-json-from-csv/")
async def convert_csv_to_json(file: UploadFile = File(...)):
    try:
        # Читаем содержимое CSV-файла
        contents = await file.read()
        decoded_contents = contents.decode('utf-8')
 
        # Создаём StringIO-объект для работы с CSV
        csv_string = StringIO(decoded_contents)
 
        # Читаем CSV, чистим пробелы и конвертируем в JSON
        reader = csv.DictReader(csv_string)
        json_data = [clean_row(row) for row in reader]
    except Exception as e:
        return {"error": str(e)}
    return {"data": json_data}
 
if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=8007, reload=True)

version: '3.7'
 
services:
 
  request_headers:
    image: ${DOCKER_REGISTRY}/request-headers
    build:
      args:
        HTTP_PROXY: ${HTTP_PROXY}
        HTTPS_PROXY: ${HTTPS_PROXY}
        NO_RPOXY: ${NO_RPOXY}
      context: .
    ports:
      - 8007:8007
    volumes:
      - ./src:/home/fmp
    command:
      python main.py
    networks: # Указываем ту же сеть, что и у сервера мобильной платформы
      - backend