Skip to content

Detect

provide.foundation.file.operations.detect

Simple functional API for file operation detection.

This module provides a minimal, user-friendly API for detecting file operations from filesystem events. It hides the complexity of the underlying detector system while providing all necessary functionality.

Examples:

>>> from provide.foundation.file.operations import detect, Event
>>>
>>> # Single operation detection
>>> events = [Event(...), Event(...)]
>>> operation = detect(events)
>>> if operation:
...     print(f"{operation.type}: {operation.path}")
>>>
>>> # Multiple operations detection
>>> operations = detect_all(events)
>>> for op in operations:
...     print(f"{op.type}: {op.path}")

Classes

Functions

create_detector 

create_detector(
    config: DetectorConfig | None = None,
) -> OperationDetector

Create a new operation detector instance.

Use this when you need a persistent detector for streaming detection or want custom configuration.

Parameters:

Name Type Description Default
config DetectorConfig | None

Optional detector configuration

 None

Returns:

Type Description
OperationDetector

New OperationDetector instance

Examples:

>>> from provide.foundation.file.operations import create_detector, DetectorConfig
>>>
>>> # Custom configuration
>>> config = DetectorConfig(time_window_ms=1000, min_confidence=0.8)
>>> detector = create_detector(config)
>>>
>>> # Use for streaming
>>> for event in events:
...     operation = detector.detect_streaming(event)
Source code in provide/foundation/file/operations/detect.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def create_detector(config: DetectorConfig | None = None) -> OperationDetector:
    """Create a new operation detector instance.

    Use this when you need a persistent detector for streaming detection
    or want custom configuration.

    Args:
        config: Optional detector configuration

    Returns:
        New OperationDetector instance

    Examples:
        >>> from provide.foundation.file.operations import create_detector, DetectorConfig
        >>>
        >>> # Custom configuration
        >>> config = DetectorConfig(time_window_ms=1000, min_confidence=0.8)
        >>> detector = create_detector(config)
        >>>
        >>> # Use for streaming
        >>> for event in events:
        ...     operation = detector.detect_streaming(event)
    """
    return OperationDetector(config)

detect 

detect(events: FileEvent) -> FileOperation | None

detect(events: list[FileEvent]) -> list[FileOperation]

detect(
    events: FileEvent | list[FileEvent],
    config: DetectorConfig | None = None,
) -> FileOperation | list[FileOperation] | None

Detect file operations from event(s).

This is the primary API for operation detection. It automatically determines whether to return a single operation or a list based on the input type.

Parameters:

Name Type Description Default
events FileEvent | list[FileEvent]

Single event or list of events to analyze

 required
config DetectorConfig | None

Optional detector configuration (uses defaults if not provided)

 None

Returns:

Type Description
FileOperation | list[FileOperation] | None
  • If single event provided: FileOperation | None
FileOperation | list[FileOperation] | None
  • If list provided: list[FileOperation] (may be empty)

Examples:

>>> # Single event
>>> operation = detect(event)
>>> if operation:
...     print(f"Found: {operation.operation_type}")
>>>
>>> # Multiple events
>>> operations = detect(event_list)
>>> print(f"Found {len(operations)} operations")
Source code in provide/foundation/file/operations/detect.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def detect(
    events: FileEvent | list[FileEvent], config: DetectorConfig | None = None
) -> FileOperation | list[FileOperation] | None:
    """Detect file operations from event(s).

    This is the primary API for operation detection. It automatically determines
    whether to return a single operation or a list based on the input type.

    Args:
        events: Single event or list of events to analyze
        config: Optional detector configuration (uses defaults if not provided)

    Returns:
        - If single event provided: FileOperation | None
        - If list provided: list[FileOperation] (may be empty)

    Examples:
        >>> # Single event
        >>> operation = detect(event)
        >>> if operation:
        ...     print(f"Found: {operation.operation_type}")
        >>>
        >>> # Multiple events
        >>> operations = detect(event_list)
        >>> print(f"Found {len(operations)} operations")
    """
    # Create detector (use cached default or new one with custom config)
    detector = _get_default_detector() if config is None else OperationDetector(config)

    # Handle single event
    if isinstance(events, FileEvent):
        results = detector.detect([events])
        return results[0] if results else None

    # Handle list of events
    return detector.detect(events)

detect_all 

detect_all(
    events: list[FileEvent],
    config: DetectorConfig | None = None,
) -> list[FileOperation]

Detect all operations from a list of events.

Explicit function for when you always want a list result, even for single events.

Parameters:

Name Type Description Default
events list[FileEvent]

List of events to analyze

 required
config DetectorConfig | None

Optional detector configuration

 None

Returns:

Type Description
list[FileOperation]

List of detected operations (may be empty)

Examples:

>>> operations = detect_all(events)
>>> for op in operations:
...     print(f"{op.operation_type}: {op.primary_path}")
Source code in provide/foundation/file/operations/detect.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def detect_all(events: list[FileEvent], config: DetectorConfig | None = None) -> list[FileOperation]:
    """Detect all operations from a list of events.

    Explicit function for when you always want a list result, even for single events.

    Args:
        events: List of events to analyze
        config: Optional detector configuration

    Returns:
        List of detected operations (may be empty)

    Examples:
        >>> operations = detect_all(events)
        >>> for op in operations:
        ...     print(f"{op.operation_type}: {op.primary_path}")
    """
    detector = _get_default_detector() if config is None else OperationDetector(config)
    return detector.detect(events)

detect_streaming 

detect_streaming(
    event: FileEvent,
    detector: OperationDetector | None = None,
) -> FileOperation | None

Process a single event in streaming mode.

For real-time detection, use this with a persistent OperationDetector instance. Operations are returned when patterns are detected based on time windows.

Parameters:

Name Type Description Default
event FileEvent

Single file event to process

 required
detector OperationDetector | None

Optional persistent detector instance (required for stateful detection)

 None

Returns:

Type Description
FileOperation | None

Completed operation if detected, None otherwise

Examples:

>>> # Create persistent detector for streaming
>>> from provide.foundation.file.operations import OperationDetector
>>> detector = OperationDetector()
>>>
>>> # Feed events as they arrive
>>> for event in event_stream:
...     operation = detect_streaming(event, detector)
...     if operation:
...         print(f"Operation detected: {operation.operation_type}")
>>>
>>> # Flush at end
>>> remaining = detector.flush()
Note

This is a lower-level API. For most use cases, the batch detect() function is simpler and sufficient.

Source code in provide/foundation/file/operations/detect.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def detect_streaming(
    event: FileEvent,
    detector: OperationDetector | None = None,
) -> FileOperation | None:
    """Process a single event in streaming mode.

    For real-time detection, use this with a persistent OperationDetector instance.
    Operations are returned when patterns are detected based on time windows.

    Args:
        event: Single file event to process
        detector: Optional persistent detector instance (required for stateful detection)

    Returns:
        Completed operation if detected, None otherwise

    Examples:
        >>> # Create persistent detector for streaming
        >>> from provide.foundation.file.operations import OperationDetector
        >>> detector = OperationDetector()
        >>>
        >>> # Feed events as they arrive
        >>> for event in event_stream:
        ...     operation = detect_streaming(event, detector)
        ...     if operation:
        ...         print(f"Operation detected: {operation.operation_type}")
        >>>
        >>> # Flush at end
        >>> remaining = detector.flush()

    Note:
        This is a lower-level API. For most use cases, the batch `detect()` function
        is simpler and sufficient.
    """
    if detector is None:
        detector = _get_default_detector()

    return detector.detect_streaming(event)