Python sqlite3 e simultaneidade

Eu tenho um programa Python que usa o módulo "threading". Uma vez a cada segundo, meu programa inicia um novo thread que busca alguns dados da web e os armazena em meu disco rígido. Gostaria de usar o sqlite3 para armazenar esses resultados, mas não consigo fazer funcionar. O problema parece ser sobre a seguinte linha:

conn = sqlite3.connect("mydatabase.db")

• Se eu colocar essa linha de código dentro de cada thread, recebo um OperationalError informando que o arquivo de banco de dados está bloqueado. Acho que isso significa que outro segmento abriu mydatabase.db por meio de uma conexão sqlite3 e o bloqueou.
• Se eu colocar esta linha de código no programa principal e passar o objeto de conexão (conn) para cada thread, recebo um ProgrammingError, dizendo que os objetos SQLite criados em uma thread só podem ser usados ​​nesse mesmo thread.

Anteriormente, eu armazenava todos os meus resultados em arquivos CSV e não tinha nenhum desses problemas de bloqueio de arquivo. Esperançosamente, isso será possível com sqlite. Alguma ideia?

Você pode usar o padrão consumidor-produtor. Por exemplo, você pode criar uma fila que é compartilhada entre threads. O primeiro thread que busca dados da web enfileira esses dados na fila compartilhada. Outro thread que possui a conexão com o banco de dados retira os dados da fila e os passa para o banco de dados.

Ao contrário da crença popular, versões mais recentes do sqlite3 fazer o acesso apoio de vários segmentos.

Isso pode ser ativado por meio de um argumento de palavra-chave opcional check_same_thread:

sqlite.connect(":memory:", check_same_thread=False)

O seguinte encontrado em mail.python.org.pipermail.1239789

Eu encontrei a solução. Não sei por que a documentação do python não contém uma única palavra sobre essa opção. Portanto, temos que adicionar um novo argumento de palavra-chave à função de conexão e seremos capazes de criar cursores a partir dele em diferentes threads. Então use:

sqlite.connect(":memory:", check_same_thread = False)

funciona perfeitamente para mim. É claro que de agora em diante eu preciso cuidar do acesso seguro de multithreading ao banco de dados. De qualquer forma, obrigado por tentar ajudar.

Mude para multiprocessamento . É muito melhor, tem uma boa escalabilidade, pode ir além do uso de vários núcleos usando várias CPUs e a interface é a mesma do módulo de threading python.

Ou, como Ali sugeriu, apenas use o mecanismo de pool de threads do SQLAlchemy . Ele vai cuidar de tudo para você automaticamente e tem muitos recursos extras, apenas para citar alguns deles:

1. SQLAlchemy inclui dialetos para SQLite, Postgres, MySQL, Oracle, MS-SQL, Firebird, MaxDB, MS Access, Sybase e Informix; A IBM também lançou um driver do DB2. Portanto, você não precisa reescrever seu aplicativo se decidir deixar de usar o SQLite.
2. O sistema Unit Of Work, uma parte central do Object Relational Mapper (ORM) do SQLAlchemy, organiza operações pendentes de criação / inserção / atualização / exclusão em filas e libera todas em um lote. Para conseguir isso, ele executa uma "classificação de dependência" topológica de todos os itens modificados na fila de modo a respeitar as restrições de chave estrangeira e agrupa instruções redundantes onde às vezes podem ser agrupados ainda mais. Isso produz o máximo de eficiência e segurança da transação e minimiza as chances de deadlocks.

Você não deveria usar threads para isso. Esta é uma tarefa trivial para torcida e que provavelmente o levaria muito mais longe.

Use apenas um thread e faça com que a conclusão da solicitação acione um evento para fazer a gravação.

twisted cuidará do agendamento, callbacks, etc ... para você. Ele vai te entregar o resultado inteiro como uma string, ou você pode executá-lo por meio de um processador de fluxo (eu tenho uma API do Twitter e uma API do friendfeed que disparam eventos para os chamadores enquanto os resultados ainda estão sendo baixados).

Dependendo do que você está fazendo com seus dados, você pode simplesmente despejar o resultado completo no sqlite quando estiver completo, cozinhá-lo e despejá-lo, ou cozinhá-lo enquanto está sendo lido e despejá-lo no final.

Eu tenho um aplicativo muito simples que faz algo parecido com o que você quer no github. Eu chamo isso de pfetch (busca paralela). Ele pega várias páginas em uma programação, transmite os resultados para um arquivo e, opcionalmente, executa um script após a conclusão bem-sucedida de cada uma. Ele também faz algumas coisas sofisticadas como GETs condicionais, mas ainda pode ser uma boa base para o que quer que você esteja fazendo.

Ou se você for preguiçoso, como eu, você pode usar SQLAlchemy . Ele tratará do threading para você ( usando thread local e algum pool de conexão ) e a maneira como o faz é até configurável .

Para um bônus adicional, se / quando você perceber / decidir que usar o Sqlite para qualquer aplicativo simultâneo será um desastre, você não terá que alterar seu código para usar MySQL, Postgres ou qualquer outra coisa. Você pode simplesmente mudar.

Você precisa usar session.close()depois de cada transação para o banco de dados para usar o mesmo cursor no mesmo thread, não usando o mesmo cursor em multi-threads que causam este erro.

Use threading.Lock ()

Eu gosto da resposta de Evgeny - as filas são geralmente a melhor maneira de implementar a comunicação entre threads. Para completar, aqui estão algumas outras opções:

• Feche a conexão do banco de dados quando os threads gerados terminarem de usá-la. Isso resolveria seu problema OperationalError, mas abrir e fechar conexões como essa geralmente é um problema, devido à sobrecarga de desempenho.
• Não use tópicos filhos. Se a tarefa de uma vez por segundo for razoavelmente leve, você pode se dar ao luxo de buscar e armazenar, e então dormir até o momento certo. Isso é indesejável, pois as operações de busca e armazenamento podem levar> 1 segundo, e você perde o benefício dos recursos multiplexados que possui com uma abordagem multi-thread.

Você precisa projetar a simultaneidade para o seu programa. O SQLite tem limitações claras e você precisa obedecê-las, consulte o FAQ (também a pergunta a seguir).

Scrapy parece uma resposta potencial à minha pergunta. Sua página inicial descreve exatamente minha tarefa. (Embora eu não tenha certeza de quão estável é o código ainda.)

Eu daria uma olhada no módulo y_serial Python para persistência de dados: http://yserial.sourceforge.net

que lida com problemas de deadlock em torno de um único banco de dados SQLite. Se a demanda por simultaneidade ficar pesada, pode-se facilmente configurar a classe Farm de muitos bancos de dados para difundir a carga ao longo do tempo estocástico.

Espero que isso ajude seu projeto ... deve ser simples o suficiente para implementar em 10 minutos.

Não consegui encontrar benchmarks em nenhuma das respostas acima, então escrevi um teste para fazer benchmarks de tudo.

Eu tentei 3 abordagens

1. Ler e escrever sequencialmente a partir do banco de dados SQLite
2. Usando um ThreadPoolExecutor para ler / escrever
3. Usando um ProcessPoolExecutor para ler / escrever

Os resultados e conclusões do benchmark são os seguintes

1. Leituras / gravações sequenciais funcionam melhor
2. Se você deve processar em paralelo, use o ProcessPoolExecutor para ler em paralelo
3. Não execute nenhuma gravação usando ThreadPoolExecutor ou ProcessPoolExecutor, pois você encontrará erros de banco de dados bloqueados e terá que tentar inserir o trecho novamente

Você pode encontrar o código e a solução completa para os benchmarks em minha resposta SO AQUI Espero que ajude!

O motivo mais provável para obter erros com bancos de dados bloqueados é que você deve emitir

conn.commit()

depois de terminar uma operação de banco de dados. Caso contrário, seu banco de dados será bloqueado contra gravação e permanecerá assim. Os outros encadeamentos que estão aguardando para gravar expirarão após um tempo (o padrão é definido como 5 segundos, consulte http://docs.python.org/2/library/sqlite3.html#sqlite3.connect para obter detalhes sobre isso) .

Um exemplo de uma inserção correta e simultânea seria este:

import threading, sqlite3
class InsertionThread(threading.Thread):

    def __init__(self, number):
        super(InsertionThread, self).__init__()
        self.number = number

    def run(self):
        conn = sqlite3.connect('yourdb.db', timeout=5)
        conn.execute('CREATE TABLE IF NOT EXISTS threadcount (threadnum, count);')
        conn.commit()

        for i in range(1000):
            conn.execute("INSERT INTO threadcount VALUES (?, ?);", (self.number, i))
            conn.commit()

# create as many of these as you wish
# but be careful to set the timeout value appropriately: thread switching in
# python takes some time
for i in range(2):
    t = InsertionThread(i)
    t.start()

Se você gosta de SQLite, ou tem outras ferramentas que funcionam com bancos de dados SQLite, ou deseja substituir arquivos CSV por arquivos db SQLite, ou precisa fazer algo raro como IPC interplataforma, então o SQLite é uma ótima ferramenta e muito adequada para o propósito. Não se deixe pressionar a usar uma solução diferente se ela não parecer certa!