Как создать API для получения мобильных контактов и хранения их в пользовательских таблицах в базе данных? ⇐ Python

[Anonymous]

Я создаю серверный API, используя FastAPI и SQLAlchemy. Цель состоит в том, чтобы:
получить контакты с мобильного устройства (с помощью внешнего интерфейса или метода на основе приложения).
отправить контакты на серверную часть.
сохранить контакты в базе данных, где контакты каждого пользователя сохраняются в отдельной таблице, динамически создаваемой на основе идентификатора пользователя.
Рабочий процесс
Фронтенд:
Запросить доступ к контактам пользователя (например, с помощью всплывающего окна).
Получите контакты с мобильного устройства.
Отправьте контакты во внутренний API вместе с идентификатором пользователя.
Бэкенд:
Динамически создать таблицу для хранения контактов, соответствующую идентификатору пользователя (например, contact_user_).
Вставьте контакты во вновь созданную таблицу.
Убедитесь, что дублирующиеся контакты отсутствуют. не создается, если таблица уже существует.
Вот моя текущая попытка:
Динамическое создание таблицы
from sqlalchemy import Table, Column, Integer, String, MetaData
metadata = MetaData()

class UserCreateService:
def create_user_contact_table(self, user_id: int):
"""
Dynamically create a table for storing contacts for a specific user.
"""
table_name = f"contacts_user_{user_id}"
return Table(
table_name,
metadata,
Column("id", Integer, primary_key=True, autoincrement=True),
Column("name", String, nullable=True),
Column("phone_number", String, nullable=False),
Column("email", String, nullable=True),
)

Сохранить конечную точку контактов
from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession

@auth_router.post("/save_contacts")
async def save_contacts_dynamic(
request: SaveContactsRequest,
db: AsyncSession = Depends(get_db)
):
"""
Save fetched contacts to a dynamically created table based on the user's ID.
"""
response = copy.deepcopy(api_response)
try:
# Dynamically create the table for the user
contact_table = UserCreateService().create_user_contact_table(request.user_id)

# Use a ThreadPoolExecutor to execute the table creation synchronously
def create_table():
sync_engine = create_engine(str(db.bind.url)) # Create a sync engine
with sync_engine.connect() as connection:
contact_table.create(bind=connection, checkfirst=True)

executor = ThreadPoolExecutor(max_workers=1)
await asyncio.get_event_loop().run_in_executor(executor, create_table)

# Insert contacts into the dynamic table
contacts_to_insert = [
{"name": contact.name, "phone_number": contact.phone_number, "email": contact.email}
for contact in request.contacts
]
await db.execute(contact_table.insert().values(contacts_to_insert))
await db.commit()

response.update({"data": [], "status": True, "message": "Contacts saved successfully!"})
except Exception as e:
logger.exception(msg=f"Input: {str(locals())}\nOutput: {str(response)} \nException : {str(e)}", exc_info=True)
error = ExceptionHandling(e=str(e), function_name="api: save_contacts").exception_handling()
response.update({"error": error, "status": False, "message": "We are having an issue for this request, please try again."})

return JSONResponse(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content=response)

return JSONResponse(status_code=status.HTTP_200_OK, content=response)


В этом коде я получаю ошибку типа
{
"data": {},
"error": {
"class-method": "api: save_contacts",
"exception_msg": "greenlet_spawn has not been called; can't call await_only() here. Was IO attempted in an unexpected place? (Background on this error at: https://sqlalche.me/e/20/xd2s)"
},
"message": "We are having an issue for this request, please try again.",
"status": false
}


Подробнее здесь: https://stackoverflow.com/questions/793 ... user-speci