pydux-test/project.py

# Import dependencies
import pydux
import asyncio
import time
import pydux_thunk
import pydux_sqlite

# Import modules
import counter
import calculate
import instrument_manager

# Define the async loop
loop = asyncio.get_event_loop()
pydux_thunk.loop = loop

pydux_sqlite.thread.start()

# Combine the middleware
middleware = [
    pydux_thunk.middleware,
    pydux_sqlite.middleware
]
middleware = pydux.apply_middleware(*middleware)

# Combine the state
initial_state = {
    'counter': counter.initial_state,
    'calculate': calculate.initial_state,
    'instrument_manager': instrument_manager.initial_state
}

# Combine the reducers
root_reducer = pydux.combine_reducers({
    'counter': counter.reducer,
    'calculate': calculate.reducer,
    'instrument_manager': instrument_manager.reducer
})

# Create the store
store = pydux.create_store(root_reducer, initial_state, middleware)


def my_subscriber():
    """
    Dummy subscriber, just prints the state.
    :return: Nothing
    """
    print('Current state', store.get_state())

# Subscribe to the store
store.subscribe(my_subscriber)

print('starting.')
store.dispatch(counter.ActionCreators.increment())
store.dispatch(counter.ActionCreators.increment())
store.dispatch(counter.ActionCreators.increment())
store.dispatch(calculate.ActionCreators.calculate(3, 4))
time.sleep(1)
store.dispatch(calculate.ActionCreators.calculate(5, 0))
time.sleep(1)
store.dispatch(calculate.ActionCreators.calculate(9, 3))
store.dispatch(counter.ActionCreators.increment())
store.dispatch(counter.ActionCreators.increment())
store.dispatch(counter.ActionCreators.increment())
store.dispatch(counter.ActionCreators.increment())
loop.run_forever()
pydux_sqlite.thread.join()