Neste post, mostro como vigiar um diretório e fazer algo quando um arquivo é criado, modificado, movido, excluido ou renomeado.
Pre-requisitos
Para criar o vigia, precisamos de instalar o pacote watchdog. Para isso, no terminal, execute o seguinte comando:
pip install watchdog
(A versão utilizada neste post é a 0.9.0)
Como funciona
Vamos precisar de um handler, que é quem vai executar as ações em si; Um Observer, que vai ficar vigiando os diretórios e algo para manter o script ocupado, pois este monitor trabalha em uma thread paralela.
Criando o Handler
O que queremos fazer é vigiar um diretório, então precisamos importar o handler correto:
from watchdog.events import FileSystemEventHandler
Agora temos que criar uma classe que herde deste handler e implementar as ações que serão utilizadas. Vou fazer dois handlers: Um com métodos específicos e outro genérico. (Calma, tudo vai fazer sentido).
Handler 1: Genérico
class WatchdogHandlerAny(FileSystemEventHandler): def on_any_event(self, event): """Will be triggered if anything happen in the folder we're watching.""" print("Something happened!") print(f"Event type: {event.event_type}") print(event)
Na classe criada acima, o método on_any_event é disparado sempre que qualquer coisa acontecer dentro da pasta que estamos vigiando.
Handler 2: Específico
class WatchdogHandler(FileSystemEventHandler): def on_modified(self, event): """Triggered when something is modified.""" print(f"File modified: {event.src_path}") def on_created(self, event): """Triggered when a file or folder is created.""" print(f"File created: {event.src_path}") def on_deleted(self, event): """Triggered when a file or folder is removed.""" print(f"File deleted: {event.src_path}") def on_moved(self, event): """Triggered when a file or folder is moved. Note: Renaming is considered moving. Will only be considered moving if it's between watched folders.""" print(f"File moved: From: {event.src_path} | To: {event.dest_path")
Este segundo handler possui uma classe separada para cada um dos eventos que podem ocorrer:
- Criação de arquivo/pasta;
- Exclusão de arquivo/pasta;
- Alteração de arquivo/pasta;
- Quando um arquivo/diretório é movido de um lugar para outro dentro das pastas vigiada.
- Um lembrete: Renomear um arquivo ou diretório é o equivalente a mover. Este não é o funcionamento do pacote, é assim que o arquivo de sistemas funciona.
Criando o Observer
class FolderWatchDog: def __init__(self, handler, watch_path): """ Constructor :param handler: Who will handle what happens in our watch_path. """ self.handler = handler self.handler = watch_path self.observer = Observer() # This is the class that does the actual watching. def start(self): self.observer.schedule(self.handler, path=self.watch_patch, recursive=True) # If set to true, will watch every folder recursively. try: print(f"Starting to watch folder: {self.watch_patch}") self.observer.start() while True: sleep(5) except KeyboardInterrupt: self.stop() except Exception as e: print(f"Something went wrong...") print(e) finally: print("All done!") def stop(self): self.observer.stop() self.observer.join()
Para deixar o código encapsulado, preferi criar uma classe. Todavia, você não tem que fazer isso, pode utilizar apenas o objeto Observer (com os métodos schedule, start, stop, e join).
Explicando a classe acima:
- Constructor:
- Recebe o caminho que será vigiado (watch_path), a instancia do handler e o observer é instanciado;
- Método start:
- A primeira coisa é chamar o método schedule do observer. Nele definimos o handler, o caminho que será vigiado e se apenas o diretório informado será vigiado ou se todos os diretórios a partir dele serão vigiados recursivamente.
- (recursive=True significa que todos os diretórios a partir do caminho informado vão ser vigiados)
- Depois chamo o método start do observer. A partir deste momento, os caminhos estarão sendo vigiados.
- O próximo passo é manter o script ocupado, para que ele não encerre a execução. Isso é feito através do while True com sleep.
- Quando for para encerrar o monitoramento, o usuário vai apertar Ctrl+C e isso vai causar um KeyboardInterrupt exception. Neste catch eu chamo o método stop (descrito no item 3 desta lista)
- A primeira coisa é chamar o método schedule do observer. Nele definimos o handler, o caminho que será vigiado e se apenas o diretório informado será vigiado ou se todos os diretórios a partir dele serão vigiados recursivamente.
- Método stop:
- Dentro do método stop da classe que criei para encapsular o Observer, chamo o método stop dele e depois join. Encerrando a thread e o parando de vigiar os diretórios.
O próximo passo é executar este código.
Iniciando o Watchdog
if __name__ == '__main__': to_watch = "/path/to/watch" w_dog = FolderWatchDog(handler=WatchdogHandlerAny()) w_dog.start()
Para executar é bem simples:
- Definir o caminho que será vigiado (to_watch);
- Instanciar a classe que criei (que encapsula o Observer), passando no construtor uma instancia do handler 1 ou 2
- Chamar o método start da classe FolderWatchDog.
- Pronto!
Se quiser, no meu Github tem um exemplo parecido com este.
Espero ter ajudado!
Latest posts by Breno RdV (see all)
- Estamos de mudança! - abril 28, 2024
- O que é Metaclass e como ela funciona. (#python #dev #metaclass) - janeiro 11, 2023
- Entenda a mágica dos Generators. (#python, #dev, #generator, #iterator) - dezembro 28, 2022