pywatchman

3.0.0 · active · verified Wed Apr 15

pywatchman is a Python client library for Facebook's Watchman file watching service. It provides a simple interface to connect to and query the Watchman daemon to efficiently discover file system changes. The library is actively maintained with frequent development tags on GitHub, and new PyPI releases (currently 3.0.0) are made periodically.

Warnings

Install

Imports

Quickstart

This quickstart demonstrates how to connect to the Watchman daemon, establish a watch on a directory, and then query for recently changed Python files within that watched directory. It includes basic error handling for when the Watchman service is not available.

import pywatchman
import os

def main():
    try:
        # Establish a connection to the Watchman daemon
        client = pywatchman.client()
        print(f"Connected to Watchman: {client.query('get-sockname')['sockname']}")

        # Define the directory to watch (current directory for example)
        watch_dir = os.getcwd()
        print(f"Attempting to watch directory: {watch_dir}")

        # Tell Watchman to start watching the directory
        # This will return existing watch if already present
        watch_result = client.query('watch', watch_dir)
        print(f"Watch established: {watch_result}")

        # Perform a query for Python files that have changed since the last clock
        # The 'clock' value is automatically managed by Watchman
        query_result = client.query(
            'query', watch_dir, {
                'expression': ['allof', ['type', 'f'], ['suffix', 'py']],
                'fields': ['name', 'size', 'mtime_ms']
            }
        )
        print("\nRecently changed Python files:")
        for f in query_result.get('files', []):
            print(f" - {f['name']} (size: {f['size']} bytes, modified: {f['mtime_ms']}ms ago)")

    except pywatchman.Unavailable as e:
        print(f"Error: Watchman service is unavailable. Please ensure Watchman is installed and running. Details: {e}")
    except Exception as e:
        print(f"An unexpected error occurred: {e}")

if __name__ == '__main__':
    main()

view raw JSON →