Nieblokujący rurociąg Scrapy do bazy danych

Question

Mam skrobaczkę internetową w aplikacji Scrapy, która pobiera elementy danych. Chcę również asynchronicznie wstawiać je do bazy danych.

Na przykład, mam transakcję, która wstawia pewne elementy do mojego db przy użyciu sqlalchemy Rdzeń:

def process_item(self, item, spider): 
    with self.connection.begin() as conn: 
     conn.execute(insert(table1).values(item['part1']) 
     conn.execute(insert(table2).values(item['part2']) 

Rozumiem, że jest to możliwe do wykorzystania sqlalchemy Rdzeń asynchronicznie z wykrzywioną alchimia. Poniżej znajduje się przykład kodu dokumentacji dla alchimia.

Czego nie rozumiem, to w jaki sposób mogę użyć powyższego kodu w strukturze alchimia. Jak skonfigurować process_item do korzystania z reaktora?

Czy mogę zrobić coś takiego?

@inlineCallbacks 
def process_item(self, item, spider): 
    with self.connection.begin() as conn: 
     yield conn.execute(insert(table1).values(item['part1']) 
     yield conn.execute(insert(table2).values(item['part2']) 

Jak napisać część reaktora?

Czy istnieje prostszy sposób na niezablokowanie wstawiania bazy danych w rurociągu Scrapy?

---

Dla porównania, tutaj jest przykład kodu z dokumentacji alchimia „s:

from alchimia import TWISTED_STRATEGY 

from sqlalchemy import (
    create_engine, MetaData, Table, Column, Integer, String 
) 
from sqlalchemy.schema import CreateTable 

from twisted.internet.defer import inlineCallbacks 
from twisted.internet.task import react 


@inlineCallbacks 
def main(reactor): 
    engine = create_engine(
     "sqlite://", reactor=reactor, strategy=TWISTED_STRATEGY 
    ) 

    metadata = MetaData() 
    users = Table("users", metadata, 
     Column("id", Integer(), primary_key=True), 
     Column("name", String()), 
    ) 

    # Create the table 
    yield engine.execute(CreateTable(users)) 

    # Insert some users 
    yield engine.execute(users.insert().values(name="Jeremy Goodwin")) 
    yield engine.execute(users.insert().values(name="Natalie Hurley")) 
    yield engine.execute(users.insert().values(name="Dan Rydell")) 
    yield engine.execute(users.insert().values(name="Casey McCall")) 
    yield engine.execute(users.insert().values(name="Dana Whitaker")) 

    result = yield engine.execute(users.select(users.c.name.startswith("D"))) 
    d_users = yield result.fetchall() 
    # Print out the users 
    for user in d_users: 
     print "Username: %s" % user[users.c.name] 

if __name__ == "__main__": 
    react(main, []) 

Answer 1

Jak mogę skonfigurować process_item używać reaktor?

Nie trzeba zarządzać innym reaktorem w rurociągu.
Zamiast tego można wykonać asynchroniczne interakcje z bazami danych w strumieniu elementów, zwracając odroczony przekaz z potoku.

Zobacz także Scrapy's doc i sample code doing asynchronous operations within an item pipeline by returning a deferred.