He estado usando la base de datos en tiempo real de firebase recientemente y no estoy seguro de por qué cuando creo una referencia de base de datos en python, simplemente no puedo obtener los datos en esa ubicación. En IPython, puedo hacer lo siguiente con éxito:

 from firebase_admin import credentials, auth, initialize_app, db cred = credentials.Certificate("my_certificate.json") initialize_app(cred, {'databaseURL':'link_to_my_database'}) ref = db.reference("some_text_here") print(ref.get()) >>> desired output

Pero cuando coloco este código en mi código API REST (usando FastAPI), obtengo un valor "Ninguno". Sé que firebase para python es una E/S de bloqueo, pero no entiendo cómo eso podría afectar mi código si todo funciona en IPython. Realmente agradecería cualquier ayuda para mostrarme dónde necesito arreglar mi código. No estoy seguro de si debería simplemente cambiar a Node.js debido a este bloqueo de E/S con Python

Actualización (28 de junio de 2020)

Basado en un comentario sobre la pregunta, agregué mi código FastAPI a continuación para mostrar dónde tengo problemas:

Enrutador.py

 import asyncio from app.controllers.pyController import PyController async def new_func(some_data: str, udata:dict): eventLoop = asyncio.get_event_loop() try: eventLoop.run_until_complete(PyController().pyController_func(some_data, udata, eventLoop)) finally: print("maybe we're done?")

pyController.py

 import redis from app.services.runPyService import RunPyService class PyController: async def pyController_func(self, some_data, uData, eventLoop): r = redis.Redis() dStoreVal = r.hget(some_data, "data") done = await RunPyService().runPyService_func(dStoreVal, uData, eventLoop)

runPyService.py

 from firebase_admin import db import asyncio import concurrent class RunPyService: executor = concurrent.futures.ThreadPoolExecutor(max_workers=40) async def runPyService_func(self, dStoreVal, uData, eventLoop): coroutine = [self.getData(dStoreVal, eventLoop)] completed, pending = await asyncio.wait(coroutine) for item in completed: print(item.result()) async def getData(self, dStoreVal, eventLoop): ref = db.reference("where-data-is-located") return await eventLoop.run_in_executor(self.executor, ref.get)

Estoy pasando el eventLoop que obtengo del enrutador a través del controlador al servicio donde realmente lo uso. Por lo general, en este punto de mi código aparece un error que dice TypeError: 'NoneType' object is not callable . Esto generalmente se debe a que getData() no devuelve ninguno. Ref.get() nunca obtiene los datos en esa ubicación.