I try to write code that will send (produce) and get (consume) some requests/responses using external service. Idea is to control all states (waiting in queue, producing, consuming, error, status check, verification and etc), map them to database and save to understand states of processes. Problem is in commiting automatic transit changes of states occured internally by itself.
I'm using sqlalchemy orm for mapping my model class with postgres database.
Example of model:
class RunProcess(Base):
__tablename__ = "process"
process_id = Column(String, primary_key=True, nullable=False)
volume = Column(Integer)
time_start = Column(DateTime)
state = Column(String(40))
time_end = Column(DateTime)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.process_id = str(uuid.uuid4())
def __repr__(self):
return f'id: {self.process_id}, start time: {self.time_start}, ' \
f'end time {self.time_end}'
I inherit from this model class to construct context class of state pattern in python:
class Run(RunProcess):
_state = None
def __init__(self, data: list,
*args, **kwargs) -> None:
self.data = data
local_time = time.time()
time_start = datetime.datetime.fromtimestamp(local_time)
super().__init__(time_start=time_start,
*args, **kwargs)
self.transition_to_state(WaitingInQueue())
def transition_to_state(self, state: State, *args, **kwargs) -> None:
self._state = state
self._state.context = self
# State name
self.state = state.__class__.__name__
def request(self, *args, **kwargs):
self._state.handle()
def __str__(self):
return f"State: {self._state}"
And my defined states:
# <<Interface>>
class State(ABC):
@property
def context(self) -> Run:
return self._context
@context.setter
def context(self, context: Run) -> None:
self._context = context
@abstractmethod
def handle(self, *args, **kwargs) -> None:
pass
def __repr__(self) -> str:
return str(self.__class__.__name__)
# Derived classes for states:
# WaitingInQueue is initial state,
# it's assigned as soon as created instance of Run class (run = Run([])).
#
# And after making request() it transit to state Producing
# that will automatically start handle().
class WaitingInQueue(State):
def handle(self, *args, **kwargs) -> None:
print("Wait")
# ... Do some business logic ...
self._context.transition_to_state(Producing())
self._context._state.handle()
# Producing is second state, it's assigned as soon as you
# use run.request() in WaitingInQueue state.
#
# It's started itself and automatically transit it state to
# Consuming without waiting from instance request() usage
class Producing(State):
def handle(self, *args, **kwargs) -> None:
print('Produce', self._context.data)
time.sleep(60)
self._produce()
def _produce(self, *args, **kwargs):
# ... Here some business logic and make automatic transition after it ...
self._context.transition_to_state(Consuming())
class Consuming(State):
pass
The main limitation is in idle in transaction only 15 minutes.
Producing and consuming can be very long, that's why creating session = Session() without context or controlling commit and rollback manually inside my Run class instance can lead to error in some time (IdleInTransactionSessionTimeout). Moreover, if I do so, my instance become not available after inner commit, e.g.:
I add Session:
class Run(RunProcess):
def __init__(self, data: list, Session: Session, # here
*args, **kwargs) -> None:
self.data = data
self.Session = Session # and here
...
def transition_to_state(self, state: State, *args, **kwargs) -> None:
...
with self.Session() as session:
session.add(self)
session.commit()
...
And when I try to use such construction:
>>> from run import Run
>>> from models.database import Session
# Make instance with predefined Session from factory
>>> run = Run([1,2,3], Session=Session)
# Trying to get __repr__ from RunProcess that was inherited from
>>> run
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "/home/ubuntu/d.studnikov/work/Projects/data_collection/smev/src/smev/smev3/database.py", line 202, in __repr__
return f'id: {self.process_id}, start time: {self.time_start}, ' \
File "/home/ubuntu/d.studnikov/work/Projects/data_collection/smev/src/smev/.venv/lib/python3.8/site-packages/sqlalchemy/orm/attributes.py", line 481, in __get__
return self.impl.get(state, dict_)
File "/home/ubuntu/d.studnikov/work/Projects/data_collection/smev/src/smev/.venv/lib/python3.8/site-packages/sqlalchemy/orm/attributes.py", line 941, in get
value = self._fire_loader_callables(state, key, passive)
File "/home/ubuntu/d.studnikov/work/Projects/data_collection/smev/src/smev/.venv/lib/python3.8/site-packages/sqlalchemy/orm/attributes.py", line 972, in _fire_loader_callables
return state._load_expired(state, passive)
File "/home/ubuntu/d.studnikov/work/Projects/data_collection/smev/src/smev/.venv/lib/python3.8/site-packages/sqlalchemy/orm/state.py", line 710, in _load_expired
self.manager.expired_attribute_loader(self, toload, passive)
File "/home/ubuntu/d.studnikov/work/Projects/data_collection/smev/src/smev/.venv/lib/python3.8/site-packages/sqlalchemy/orm/loading.py", line 1369, in load_scalar_attributes
raise orm_exc.DetachedInstanceError(
sqlalchemy.orm.exc.DetachedInstanceError: Instance <Run at 0x7fd6c9f6b9a0> is not bound to a Session; attribute refresh operation cannot proceed (Background on this error at: https://sqlalche.me/e/14/bhk3)
# If we access to __str__ of our Run class it's ok
>>> print(run)
State: WaitingInQueue
I know that it is a bad practice due to breaking separate and external principle. However, I don't see any other solutions for committing inexplicit inner state changes. Furthermore, I don't have an opportunity to get attributes of my instance that were mapped by sqlalchemy orm and bounded to session... Of course, I can instance session = Session.object_session(run), make scoped_session() or expunge() and get access to all bounded attributes outside of instance, but working with two different session scopes inside and outside of instance is bad decision I think.
On the other hand, if I apply with Session() as session context and won't pass it to signature, I'll not be able to commit changes in state that were occured automatically by instance itself , e.g.:
>>> from run import Run
>>> from models.database import Session
# This is instance of Run class without Session inside
>>> run = Run([1,2,3])
>>> with Session() as session:
... session.add(run)
... session.commit()
... print(run, repr(run))
... run.get_smev()
... print(run, repr(run))
... session.commit()
...
# Output:
State: WaitingInQueue
id: 3d358344-4c9d-417c-bca6-2f56310c67ee, start time: 2022-09-06 14:30:38.984317, end time None
Wait
Produce [1, 2, 3]
State: Consuming
id: 3d358344-4c9d-417c-bca6-2f56310c67ee, start time: 2022-09-06 14:30:38.984317, end time 2022-09-06 14:33:30.49744
So here in database we will see committed only WaitingInQueue state and Consuming states, but Producing state won't be reflected and appeared in DB as run instance automatically transit to Consuming state.
I've read many answers on sqlalchemy Session issues, but I've not found solution to the problem. What should I do to commit Producing state, when it changes to other state itself, and prevent idle in transaction? Is there any way to use Session inside class, self committing changes and working with instance outside like with ordinary object in python?