I don't know why the firebase-rest-api and the pyrebase isn't closing the stream instance when KeyboardInterrupt exception is raised..
You can receive this exception in this loop code:
from firebase import initialize_app
from typing import List
class MyStuffTracker(object):
"""Tracks changes of my stuff in Firebase, this a public db for tests."""
_db = initialize_app({
'apiKey': "AIzaSyBtwpgXl4QHGsFtjvpGRGccHA31Jp8HdZ8",
'authDomain': "website-acca0.firebaseapp.com",
'databaseURL': "https://website-acca0-default-rtdb.firebaseio.com",
'projectId': "website-acca0",
'storageBucket': "website-acca0.appspot.com",
'messagingSenderId': "403032921703",
'appId': "1:403032921703:web:c1d5dec6bd02c1a8a3a6c2",
'measurementId': "G-QD36TFS9M5"
}).database()
my_stuff: List[dict] = None # In my example my data is a list of some dictionaries
@property
def is_ready(self) -> bool:
"""
Returns:
bool: True if my stuff is ready for use
"""
return self.my_stuff is not None
def stream_handler(self, message):
print("Got some update from the Firebase")
# We only care if something changed
if message["event"] in ("put", "patch"):
print("Something changed")
if message["path"] == "/":
print("Seems like a fresh data or everything have changed, just grab it!")
self.my_stuff: List[dict] = message["data"]
else:
print("Something updated somewhere, I dont't care I just want the latest snapshot of my stuff")
# Just get whole-data of my stuff and list (second) item of the pyres (that I expect to be a dict)
self.my_stuff: List[dict] = list(it.item[1] for it in self._db.get().pyres)
print(tracker.my_stuff)
def __init__(self) -> None:
"""Start tracking my stuff changes in Firebase"""
super().__init__()
self._db.stream(self.stream_handler)
tracker = MyStuffTracker()
while True:
pass
It's not closed...
If remove the while, the code is inclosable, I need to go to the task manager to close the python...
I know that I need to run the close method to close the stream, but when the keyboardinterrupt exception was raised, don't have a way to close anymore
I see that it solved:
tracker = MyStuffTracker()
try:
while True:
pass
finally:
tracker.close()
but it seans a little wrong, because if its closed out of the try, the code will enter on a loop on the same way...