Ufdjrw

Fewest number of steps to reach 200 using special calculator

Describing a chess game in a novel

Am I eligible for the Eurail Youth pass? I am 27.5 years old

In Aliens, how many people were on LV-426 before the Marines arrived​?

Existence of a celestial body big enough for early civilization to be thought of as a second moon

What does "Four-F." mean?

Violin - Can double stops be played when the strings are not next to each other?

Using Past-Perfect interchangeably with the Past Continuous

What is the English word for a graduation award?

Is it possible to stack the damage done by the Absorb Elements spell?

Is it true that good novels will automatically sell themselves on Amazon (and so on) and there is no need for one to waste time promoting?

HP P840 HDD RAID 5 many strange drive failures

Why is indicated airspeed rather than ground speed used during the takeoff roll?

Unfrosted light bulb

Asserting that Atheism and Theism are both faith based positions

Does the attack bonus from a Masterwork weapon stack with the attack bonus from Masterwork ammunition?

Is there a creature that is resistant or immune to non-magical damage other than bludgeoning, slashing, and piercing?

Worshiping one God at a time?

What is the relationship between relativity and the Doppler effect?

Recruiter wants very extensive technical details about all of my previous work

Geography in 3D perspective

Generic TVP tradeoffs?

What (if any) is the reason to buy in small local stores?

How to get the n-th line after a grepped one?

Pipelines with asyncio coroutines

3

I'm new with asyncio and I'm implemented some kind of pipeline using asyncio coroutines.
The main idea is to have different pipelines with coroutines that are connected to
aggregate information in a payload and after that save the payload in a database.

class ExamplePipeline(PipelineBaseClass):
 def __init__(self):
 PipelineBaseClass.__init__(self)
 self.stages = [get_some_info, build_message,n,..,save_to_db]
 self.wrapper = self.connect(self.stages)

 def connect(self, stages):
 def wrapper(*args, **kwargs):
 data_out = yield from stages[0](*args, **kwargs)
 for stage in stages[1:]:
 data_out = yield from stage(data_out)
 return data_out
 return wrapper

 def run(self, data_in):
 asyncio.get_event_loop().run_until_complete(self.wrapper(data_in))

The functions for each stage are coroutines suppose something like

@asyncio.coroutine
def get_some_info(payload):
 payload = call_an_endpoint
 return payload

@asyncio.coroutine
def build_message(payload):
 #do some logic
 return payload

I have a new requirement in which the build_message co routine in some cases
needs to fork the rest of the pipeline, this is that the rest of the stages
n..save_to_db must be executed x times with different payloads since
build_message.

[get_some_info, build_message -> payload 1 [n, .. , save_to_db]
 payload 2 [n, .. , save_to_db]
 payload x [n, .. , save_to_db]

So at finish I have x different messages on the db.

How can I implement my new feature in this context?

python-3.x pipeline python-asyncio coroutine

share|improve this question

edited Mar 7 at 2:25

dirn

11.3k33749

asked Mar 6 at 22:10

Hernan AcostaHernan Acosta

3829

add a comment | 

3

I'm new with asyncio and I'm implemented some kind of pipeline using asyncio coroutines.
The main idea is to have different pipelines with coroutines that are connected to
aggregate information in a payload and after that save the payload in a database.

class ExamplePipeline(PipelineBaseClass):
 def __init__(self):
 PipelineBaseClass.__init__(self)
 self.stages = [get_some_info, build_message,n,..,save_to_db]
 self.wrapper = self.connect(self.stages)

 def connect(self, stages):
 def wrapper(*args, **kwargs):
 data_out = yield from stages[0](*args, **kwargs)
 for stage in stages[1:]:
 data_out = yield from stage(data_out)
 return data_out
 return wrapper

 def run(self, data_in):
 asyncio.get_event_loop().run_until_complete(self.wrapper(data_in))

The functions for each stage are coroutines suppose something like

@asyncio.coroutine
def get_some_info(payload):
 payload = call_an_endpoint
 return payload

@asyncio.coroutine
def build_message(payload):
 #do some logic
 return payload

I have a new requirement in which the build_message co routine in some cases
needs to fork the rest of the pipeline, this is that the rest of the stages
n..save_to_db must be executed x times with different payloads since
build_message.

[get_some_info, build_message -> payload 1 [n, .. , save_to_db]
 payload 2 [n, .. , save_to_db]
 payload x [n, .. , save_to_db]

So at finish I have x different messages on the db.

How can I implement my new feature in this context?

python-3.x pipeline python-asyncio coroutine

share|improve this question

edited Mar 7 at 2:25

dirn

11.3k33749

asked Mar 6 at 22:10

Hernan AcostaHernan Acosta

3829

add a comment | 

I'm new with asyncio and I'm implemented some kind of pipeline using asyncio coroutines.
The main idea is to have different pipelines with coroutines that are connected to
aggregate information in a payload and after that save the payload in a database.

class ExamplePipeline(PipelineBaseClass):
 def __init__(self):
 PipelineBaseClass.__init__(self)
 self.stages = [get_some_info, build_message,n,..,save_to_db]
 self.wrapper = self.connect(self.stages)

 def connect(self, stages):
 def wrapper(*args, **kwargs):
 data_out = yield from stages[0](*args, **kwargs)
 for stage in stages[1:]:
 data_out = yield from stage(data_out)
 return data_out
 return wrapper

 def run(self, data_in):
 asyncio.get_event_loop().run_until_complete(self.wrapper(data_in))

The functions for each stage are coroutines suppose something like

@asyncio.coroutine
def get_some_info(payload):
 payload = call_an_endpoint
 return payload

@asyncio.coroutine
def build_message(payload):
 #do some logic
 return payload

I have a new requirement in which the build_message co routine in some cases
needs to fork the rest of the pipeline, this is that the rest of the stages
n..save_to_db must be executed x times with different payloads since
build_message.

[get_some_info, build_message -> payload 1 [n, .. , save_to_db]
 payload 2 [n, .. , save_to_db]
 payload x [n, .. , save_to_db]

So at finish I have x different messages on the db.

How can I implement my new feature in this context?

python-3.x pipeline python-asyncio coroutine

share|improve this question

edited Mar 7 at 2:25

dirn

11.3k33749

asked Mar 6 at 22:10

Hernan AcostaHernan Acosta

3829

class ExamplePipeline(PipelineBaseClass):
 def __init__(self):
 PipelineBaseClass.__init__(self)
 self.stages = [get_some_info, build_message,n,..,save_to_db]
 self.wrapper = self.connect(self.stages)

 def connect(self, stages):
 def wrapper(*args, **kwargs):
 data_out = yield from stages[0](*args, **kwargs)
 for stage in stages[1:]:
 data_out = yield from stage(data_out)
 return data_out
 return wrapper

 def run(self, data_in):
 asyncio.get_event_loop().run_until_complete(self.wrapper(data_in))

@asyncio.coroutine
def get_some_info(payload):
 payload = call_an_endpoint
 return payload

@asyncio.coroutine
def build_message(payload):
 #do some logic
 return payload

[get_some_info, build_message -> payload 1 [n, .. , save_to_db]
 payload 2 [n, .. , save_to_db]
 payload x [n, .. , save_to_db]

share|improve this question

edited Mar 7 at 2:25

dirn

11.3k33749

asked Mar 6 at 22:10

Hernan AcostaHernan Acosta

3829

share|improve this question

edited Mar 7 at 2:25

dirn

11.3k33749

asked Mar 6 at 22:10

Hernan AcostaHernan Acosta

3829

edited Mar 7 at 2:25

dirn

11.3k33749

edited Mar 7 at 2:25

dirn

11.3k33749

asked Mar 6 at 22:10

Hernan AcostaHernan Acosta

3829

asked Mar 6 at 22:10

Hernan AcostaHernan Acosta

3829