ONVIFService
Base class for all ONVIF service implementations.
This abstract base class provides automatic error handling and consistent API
behavior for all ONVIF services (Device,
Media, PTZ, etc.).
Version History
- Available since
>=v0.1.8.
All service classes inherit from ONVIFService to ensure
- Consistent exception handling across all ONVIF operations
- Automatic wrapping of errors into
ONVIFOperationException - Uniform error reporting with operation names
- Transparent method interception without explicit wrappers
Implementation Details
- Uses
__getattribute__magic method to intercept all method calls - Automatically wraps ONVIF operations (methods starting with uppercase)
- Preserves non-ONVIF methods, private methods, and attributes
- Converts all exceptions to
ONVIFOperationExceptionfor consistency - Re-raises existing
ONVIFOperationExceptionwithout double-wrapping
Notes
- This is an abstract base class - don't instantiate directly
- Subclasses must implement their own
__init__and ONVIF methods - The
__getattribute__interception has minimal performance overhead - Error wrapping preserves full stack trace for debugging
- Compatible with all Python magic methods and properties
See Also
ONVIFOperationException: Exception class for wrapped errorsONVIFOperator: Low-level SOAP operation handler- Device, Media, PTZ, etc.: Concrete service implementations
Source code in onvif\utils\service.py
class ONVIFService:
"""Base class for all ONVIF service implementations.
This abstract base class provides automatic error handling and consistent API
behavior for all ONVIF services ([`Device`](../services/device.md),
[`Media`](../services/media.md), [`PTZ`](../services/ptz.md), etc.).
!!! tip "Version History"
- Available since [`>=v0.1.8`](/onvif-python/releases/#v0.1.8).
!!! danger "All service classes inherit from `ONVIFService` to ensure"
- Consistent exception handling across all ONVIF operations
- Automatic wrapping of errors into [`ONVIFOperationException`](../exceptions/onvif_exception.md)
- Uniform error reporting with operation names
- Transparent method interception without explicit wrappers
!!! abstract "Implementation Details"
- Uses `__getattribute__` magic method to intercept all method calls
- Automatically wraps ONVIF operations (methods starting with uppercase)
- Preserves non-ONVIF methods, private methods, and attributes
- Converts all exceptions to [`ONVIFOperationException`](../exceptions/onvif_exception.md) for consistency
- Re-raises existing [`ONVIFOperationException`](../exceptions/onvif_exception.md) without double-wrapping
??? note "Notes"
- This is an abstract base class - don't instantiate directly
- Subclasses must implement their own `__init__` and ONVIF methods
- The `__getattribute__` interception has minimal performance overhead
- Error wrapping preserves full stack trace for debugging
- Compatible with all Python magic methods and properties
??? note "See Also"
- [`ONVIFOperationException`](../exceptions/onvif_exception.md): Exception class for wrapped errors
- [`ONVIFOperator`](../core/onvif_operator.md): Low-level SOAP operation handler
- Device, Media, PTZ, etc.: Concrete service implementations
"""
def __getattribute__(self, name: str) -> Any:
"""Intercept all method calls and wrap ONVIF operations with error handling.
This magic method is called for every attribute access on the service object.
It intercepts ONVIF operation calls and wraps them with consistent error handling.
Args:
name (str): Name of the attribute or method being accessed
Returns:
The attribute value, or a wrapped method if it's an ONVIF operation
Raises:
ONVIFOperationException: If the ONVIF operation fails
!!! question "Method Interception Logic"
1. Get the attribute using object.__getattribute__
2. Skip if not callable, private, or internal attribute
3. Check if method name starts with uppercase (ONVIF convention)
4. If yes, return wrapped version that catches and converts exceptions
5. If no, return the original method as-is
"""
attr = object.__getattribute__(self, name)
# Skip non-callable attributes, private methods, and internal attributes
if not callable(attr) or name.startswith("_") or name in ["operator"]:
return attr
# Check if this looks like an ONVIF method (starts with uppercase)
if not name[0].isupper():
return attr
# Wrap ONVIF methods with error handling and Zeep object conversion
def wrapped_method(*args, **kwargs):
try:
# If called with 1 positional arg that is a Zeep object and no kwargs,
# convert the object's fields to kwargs instead of passing it as positional arg
if len(args) == 1 and not kwargs and _is_zeep_object(args[0]):
params_obj = args[0]
logger.debug("Converting Zeep object to kwargs for %s", name)
# Extract fields from Zeep object using its XSD type elements
# pylint: disable=protected-access
if hasattr(params_obj._xsd_type, "elements"):
kwargs = {}
for (
elem_name,
_,
) in params_obj._xsd_type.elements:
kwargs[elem_name] = getattr(params_obj, elem_name)
return attr(**kwargs)
logger.debug("Calling wrapped ONVIF method: %s", name)
result = attr(*args, **kwargs)
logger.debug("ONVIF method %s completed successfully", name)
return result
except ONVIFOperationException as oe:
# Re-raise ONVIF exceptions as-is
service_name = getattr(self.operator, "service_name", "Unknown")
logger.error("%s.%s: %s", service_name, name, oe)
raise
except Exception as e:
# Convert any other exception (including TypeError) to ONVIFOperationException
service_name = getattr(self.operator, "service_name", "Unknown")
logger.error("%s.%s: %s", service_name, name, e)
raise ONVIFOperationException(name, e) from e
return wrapped_method
def type(self, type_name: str) -> Any:
"""
Create and return an instance of the specified ONVIF type.
Args:
type_name (str): Name of the type to create (e.g., 'SetHostname', 'SetIPAddressFilter')
Returns:
Type instance that can be populated with data
Raises:
ONVIFOperationException: If type creation fails
Example:
```python linenums="1"
device = client.devicemgmt()
newuser = device.type('CreateUsers')
newuser.User.append({"Username": 'new_user', "Password": 'new_password', "UserLevel": 'User'})
device.CreateUsers(newuser)
hostname = device.type('SetHostname')
hostname.Name = 'NewHostname'
device.SetHostname(hostname)
time_params = device.type('SetSystemDateAndTime')
time_params.DateTimeType = 'NTP'
time_params.DaylightSavings = True
time_params.TimeZone.TZ = 'UTC+02:00'
now = datetime.now()
time_params.UTCDateTime.Date.Year = now.year
time_params.UTCDateTime.Date.Month = now.month
time_params.UTCDateTime.Date.Day = now.day
time_params.UTCDateTime.Time.Hour = now.hour
time_params.UTCDateTime.Time.Minute = now.minute
time_params.UTCDateTime.Time.Second = now.second
device.SetSystemDateAndTime(time_params)
```
"""
try:
logger.debug("Creating ONVIF type: %s", type_name)
result = self.operator.create_type(type_name)
logger.debug("Successfully created type: %s", type_name)
return result
except Exception as e:
logger.error("Failed to create type %s: %s", type_name, e)
raise ONVIFOperationException(f"type({type_name})", e) from e
def operations(self) -> list[str] | list:
"""List all available operations for this service.
Returns:
List of operation names that can be used with `type()` method
Example:
```python linenums="1"
ptz = client.ptz()
# List all available operations for each service
print("PTZ Operations:")
for op in ptz.operations():
print(f" - {op}")
# Check if specific operation is supported
if 'ContinuousMove' in ptz.operations():
print("PTZ continuous movement is supported")
```
"""
try:
# Get all methods from the service object itself (not operator.service)
# This includes all ONVIF operations that are dynamically added
operations = [
method
for method in dir(self)
if not method.startswith("_")
and method not in ["type", "desc", "operations"]
and callable(getattr(self, method))
and method[0].isupper() # ONVIF methods start with uppercase
]
# Extract service name from binding for logging context
service_name = (
self.operator.service_name
if hasattr(self.operator, "service_name")
else "Unknown"
)
logger.debug(
"Successfully listed operations for %s: %d operations found",
service_name,
len(operations),
)
return sorted(operations)
except Exception as e: # pylint: disable=broad-except
logger.error("Failed to list operations: %s", e)
return []
def desc(self, method_name: str) -> dict:
"""
Get documentation and parameter information for a specific operation/method.
Args:
method_name (str): Name of the method to describe (e.g., 'GetDeviceInformation', 'SetHostname')
Returns:
Operation documentation dictionary.
!!! abstract "Documentation dict"
| Key | Type | Description |
| --- | ---- | ----------- |
| `doc` | `str | None` | Method documentation from WSDL. |
| `required` | `list[str]` | List of required parameter names; empty if none are available. |
| `optional` | `list[str]` | List of optional parameter names; empty if none are available. |
| `method_name` | `str` | The method name. |
| `service_name` | `str` | The service name. |
Raises:
ONVIFOperationException: If method doesn't exist or documentation cannot be retrieved
Example:
```python linenums="1"
device = client.devicemgmt()
# Get description for a method
info = device.desc('GetDeviceInformation')
print(info['doc'])
print("Required params:", info['required'])
print("Optional params:", info['optional'])
# Check what methods are available first
methods = device.operations()
info = device.desc(methods[0]) # Describe first method
```
"""
try:
# Check if method exists
if not hasattr(self, method_name):
available_methods = self.operations()
raise ValueError(
f"Method '{method_name}' not found in service. "
f"Available methods: {', '.join(available_methods[:5])}{'...' if len(available_methods) > 5 else ''}"
)
# Extract service name from binding for context
service_name = (
self.operator.service_name
if hasattr(self.operator, "service_name")
else "Unknown"
)
# Always try to extract method parameters
required_args = []
optional_args = []
doc_text = None
try:
# Use object.__getattribute__ to bypass ONVIFService wrapper and get original method
method = object.__getattribute__(self, method_name)
sig = inspect.signature(method)
for param in sig.parameters.values():
if param.name != "self":
if param.default is inspect.Parameter.empty:
required_args.append(param.name)
else:
optional_args.append(param.name)
logger.debug(
"Successfully extracted parameters for %s.%s",
service_name,
method_name,
)
except Exception as param_error: # pylint: disable=broad-except
logger.warning(
"Could not extract parameters for %s: %s", method_name, param_error
)
# Try to get documentation (optional)
try:
doc_info = get_method_documentation(self, method_name)
if doc_info and doc_info.get("doc"):
# Only use doc if it's not the default "No description available" message
doc_text = doc_info["doc"]
if (
doc_text
and "No description available" not in doc_text
and "No documentation available" not in doc_text
):
# Keep the doc text as is
pass
else:
# Set to None if it's the default message
doc_text = None
except (OSError, ValueError, AttributeError) as e:
logger.warning(
"Could not retrieve documentation for %s: %s", method_name, e
) # Documentation is optional
return {
"doc": doc_text,
"required": required_args,
"optional": optional_args,
"method_name": method_name,
"service_name": service_name,
}
except Exception as e:
logger.error("Failed to get description for method %s: %s", method_name, e)
raise ONVIFOperationException(f"desc({method_name})", e) from e
def to_dict(self, zeep_object: Any) -> dict:
"""
Convert a zeep object (result from ONVIF operation) to Python dictionary.
Args:
zeep_object (Any): The zeep object returned from ONVIF operations
Returns:
Python dictionary representation of the zeep object
Example:
```python linenums="1"
device = client.devicemgmt()
info = device.GetDeviceInformation()
info_dict = device.to_dict(info)
print(info_dict)
```
"""
try:
return (
{}
if zeep_object is None
else zeep.helpers.serialize_object(zeep_object)
)
except Exception as e: # pylint: disable=broad-except
logger.error("Failed to convert zeep object to dict: %s", e)
return {}
__getattribute__(name: str) -> Any
Intercept all method calls and wrap ONVIF operations with error handling.
This magic method is called for every attribute access on the service object. It intercepts ONVIF operation calls and wraps them with consistent error handling.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Name of the attribute or method being accessed |
required |
Returns:
| Type | Description |
|---|---|
Any
|
The attribute value, or a wrapped method if it's an ONVIF operation |
Raises:
| Type | Description |
|---|---|
ONVIFOperationException
|
If the ONVIF operation fails |
Method Interception Logic
- Get the attribute using object.getattribute
- Skip if not callable, private, or internal attribute
- Check if method name starts with uppercase (ONVIF convention)
- If yes, return wrapped version that catches and converts exceptions
- If no, return the original method as-is
Source code in onvif\utils\service.py
def __getattribute__(self, name: str) -> Any:
"""Intercept all method calls and wrap ONVIF operations with error handling.
This magic method is called for every attribute access on the service object.
It intercepts ONVIF operation calls and wraps them with consistent error handling.
Args:
name (str): Name of the attribute or method being accessed
Returns:
The attribute value, or a wrapped method if it's an ONVIF operation
Raises:
ONVIFOperationException: If the ONVIF operation fails
!!! question "Method Interception Logic"
1. Get the attribute using object.__getattribute__
2. Skip if not callable, private, or internal attribute
3. Check if method name starts with uppercase (ONVIF convention)
4. If yes, return wrapped version that catches and converts exceptions
5. If no, return the original method as-is
"""
attr = object.__getattribute__(self, name)
# Skip non-callable attributes, private methods, and internal attributes
if not callable(attr) or name.startswith("_") or name in ["operator"]:
return attr
# Check if this looks like an ONVIF method (starts with uppercase)
if not name[0].isupper():
return attr
# Wrap ONVIF methods with error handling and Zeep object conversion
def wrapped_method(*args, **kwargs):
try:
# If called with 1 positional arg that is a Zeep object and no kwargs,
# convert the object's fields to kwargs instead of passing it as positional arg
if len(args) == 1 and not kwargs and _is_zeep_object(args[0]):
params_obj = args[0]
logger.debug("Converting Zeep object to kwargs for %s", name)
# Extract fields from Zeep object using its XSD type elements
# pylint: disable=protected-access
if hasattr(params_obj._xsd_type, "elements"):
kwargs = {}
for (
elem_name,
_,
) in params_obj._xsd_type.elements:
kwargs[elem_name] = getattr(params_obj, elem_name)
return attr(**kwargs)
logger.debug("Calling wrapped ONVIF method: %s", name)
result = attr(*args, **kwargs)
logger.debug("ONVIF method %s completed successfully", name)
return result
except ONVIFOperationException as oe:
# Re-raise ONVIF exceptions as-is
service_name = getattr(self.operator, "service_name", "Unknown")
logger.error("%s.%s: %s", service_name, name, oe)
raise
except Exception as e:
# Convert any other exception (including TypeError) to ONVIFOperationException
service_name = getattr(self.operator, "service_name", "Unknown")
logger.error("%s.%s: %s", service_name, name, e)
raise ONVIFOperationException(name, e) from e
return wrapped_method
type(type_name: str) -> Any
Create and return an instance of the specified ONVIF type.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
type_name
|
str
|
Name of the type to create (e.g., 'SetHostname', 'SetIPAddressFilter') |
required |
Returns:
| Type | Description |
|---|---|
Any
|
Type instance that can be populated with data |
Raises:
| Type | Description |
|---|---|
ONVIFOperationException
|
If type creation fails |
Example
device = client.devicemgmt()
newuser = device.type('CreateUsers')
newuser.User.append({"Username": 'new_user', "Password": 'new_password', "UserLevel": 'User'})
device.CreateUsers(newuser)
hostname = device.type('SetHostname')
hostname.Name = 'NewHostname'
device.SetHostname(hostname)
time_params = device.type('SetSystemDateAndTime')
time_params.DateTimeType = 'NTP'
time_params.DaylightSavings = True
time_params.TimeZone.TZ = 'UTC+02:00'
now = datetime.now()
time_params.UTCDateTime.Date.Year = now.year
time_params.UTCDateTime.Date.Month = now.month
time_params.UTCDateTime.Date.Day = now.day
time_params.UTCDateTime.Time.Hour = now.hour
time_params.UTCDateTime.Time.Minute = now.minute
time_params.UTCDateTime.Time.Second = now.second
device.SetSystemDateAndTime(time_params)
Source code in onvif\utils\service.py
def type(self, type_name: str) -> Any:
"""
Create and return an instance of the specified ONVIF type.
Args:
type_name (str): Name of the type to create (e.g., 'SetHostname', 'SetIPAddressFilter')
Returns:
Type instance that can be populated with data
Raises:
ONVIFOperationException: If type creation fails
Example:
```python linenums="1"
device = client.devicemgmt()
newuser = device.type('CreateUsers')
newuser.User.append({"Username": 'new_user', "Password": 'new_password', "UserLevel": 'User'})
device.CreateUsers(newuser)
hostname = device.type('SetHostname')
hostname.Name = 'NewHostname'
device.SetHostname(hostname)
time_params = device.type('SetSystemDateAndTime')
time_params.DateTimeType = 'NTP'
time_params.DaylightSavings = True
time_params.TimeZone.TZ = 'UTC+02:00'
now = datetime.now()
time_params.UTCDateTime.Date.Year = now.year
time_params.UTCDateTime.Date.Month = now.month
time_params.UTCDateTime.Date.Day = now.day
time_params.UTCDateTime.Time.Hour = now.hour
time_params.UTCDateTime.Time.Minute = now.minute
time_params.UTCDateTime.Time.Second = now.second
device.SetSystemDateAndTime(time_params)
```
"""
try:
logger.debug("Creating ONVIF type: %s", type_name)
result = self.operator.create_type(type_name)
logger.debug("Successfully created type: %s", type_name)
return result
except Exception as e:
logger.error("Failed to create type %s: %s", type_name, e)
raise ONVIFOperationException(f"type({type_name})", e) from e
operations() -> list[str] | list
List all available operations for this service.
Returns:
| Type | Description |
|---|---|
list[str] | list
|
List of operation names that can be used with |
Example
Source code in onvif\utils\service.py
def operations(self) -> list[str] | list:
"""List all available operations for this service.
Returns:
List of operation names that can be used with `type()` method
Example:
```python linenums="1"
ptz = client.ptz()
# List all available operations for each service
print("PTZ Operations:")
for op in ptz.operations():
print(f" - {op}")
# Check if specific operation is supported
if 'ContinuousMove' in ptz.operations():
print("PTZ continuous movement is supported")
```
"""
try:
# Get all methods from the service object itself (not operator.service)
# This includes all ONVIF operations that are dynamically added
operations = [
method
for method in dir(self)
if not method.startswith("_")
and method not in ["type", "desc", "operations"]
and callable(getattr(self, method))
and method[0].isupper() # ONVIF methods start with uppercase
]
# Extract service name from binding for logging context
service_name = (
self.operator.service_name
if hasattr(self.operator, "service_name")
else "Unknown"
)
logger.debug(
"Successfully listed operations for %s: %d operations found",
service_name,
len(operations),
)
return sorted(operations)
except Exception as e: # pylint: disable=broad-except
logger.error("Failed to list operations: %s", e)
return []
desc(method_name: str) -> dict
Get documentation and parameter information for a specific operation/method.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
method_name
|
str
|
Name of the method to describe (e.g., 'GetDeviceInformation', 'SetHostname') |
required |
Returns:
| Type | Description |
|---|---|
dict
|
Operation documentation dictionary. |
Documentation dict
| Key | Type | Description |
|---|---|---|
doc |
str | None |
Method documentation from WSDL. |
required |
list[str] |
List of required parameter names; empty if none are available. |
optional |
list[str] |
List of optional parameter names; empty if none are available. |
method_name |
str |
The method name. |
service_name |
str |
The service name. |
Raises:
| Type | Description |
|---|---|
ONVIFOperationException
|
If method doesn't exist or documentation cannot be retrieved |
Example
device = client.devicemgmt()
# Get description for a method
info = device.desc('GetDeviceInformation')
print(info['doc'])
print("Required params:", info['required'])
print("Optional params:", info['optional'])
# Check what methods are available first
methods = device.operations()
info = device.desc(methods[0]) # Describe first method
Source code in onvif\utils\service.py
def desc(self, method_name: str) -> dict:
"""
Get documentation and parameter information for a specific operation/method.
Args:
method_name (str): Name of the method to describe (e.g., 'GetDeviceInformation', 'SetHostname')
Returns:
Operation documentation dictionary.
!!! abstract "Documentation dict"
| Key | Type | Description |
| --- | ---- | ----------- |
| `doc` | `str | None` | Method documentation from WSDL. |
| `required` | `list[str]` | List of required parameter names; empty if none are available. |
| `optional` | `list[str]` | List of optional parameter names; empty if none are available. |
| `method_name` | `str` | The method name. |
| `service_name` | `str` | The service name. |
Raises:
ONVIFOperationException: If method doesn't exist or documentation cannot be retrieved
Example:
```python linenums="1"
device = client.devicemgmt()
# Get description for a method
info = device.desc('GetDeviceInformation')
print(info['doc'])
print("Required params:", info['required'])
print("Optional params:", info['optional'])
# Check what methods are available first
methods = device.operations()
info = device.desc(methods[0]) # Describe first method
```
"""
try:
# Check if method exists
if not hasattr(self, method_name):
available_methods = self.operations()
raise ValueError(
f"Method '{method_name}' not found in service. "
f"Available methods: {', '.join(available_methods[:5])}{'...' if len(available_methods) > 5 else ''}"
)
# Extract service name from binding for context
service_name = (
self.operator.service_name
if hasattr(self.operator, "service_name")
else "Unknown"
)
# Always try to extract method parameters
required_args = []
optional_args = []
doc_text = None
try:
# Use object.__getattribute__ to bypass ONVIFService wrapper and get original method
method = object.__getattribute__(self, method_name)
sig = inspect.signature(method)
for param in sig.parameters.values():
if param.name != "self":
if param.default is inspect.Parameter.empty:
required_args.append(param.name)
else:
optional_args.append(param.name)
logger.debug(
"Successfully extracted parameters for %s.%s",
service_name,
method_name,
)
except Exception as param_error: # pylint: disable=broad-except
logger.warning(
"Could not extract parameters for %s: %s", method_name, param_error
)
# Try to get documentation (optional)
try:
doc_info = get_method_documentation(self, method_name)
if doc_info and doc_info.get("doc"):
# Only use doc if it's not the default "No description available" message
doc_text = doc_info["doc"]
if (
doc_text
and "No description available" not in doc_text
and "No documentation available" not in doc_text
):
# Keep the doc text as is
pass
else:
# Set to None if it's the default message
doc_text = None
except (OSError, ValueError, AttributeError) as e:
logger.warning(
"Could not retrieve documentation for %s: %s", method_name, e
) # Documentation is optional
return {
"doc": doc_text,
"required": required_args,
"optional": optional_args,
"method_name": method_name,
"service_name": service_name,
}
except Exception as e:
logger.error("Failed to get description for method %s: %s", method_name, e)
raise ONVIFOperationException(f"desc({method_name})", e) from e
to_dict(zeep_object: Any) -> dict
Convert a zeep object (result from ONVIF operation) to Python dictionary.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
zeep_object
|
Any
|
The zeep object returned from ONVIF operations |
required |
Returns:
| Type | Description |
|---|---|
dict
|
Python dictionary representation of the zeep object |
Example
Source code in onvif\utils\service.py
def to_dict(self, zeep_object: Any) -> dict:
"""
Convert a zeep object (result from ONVIF operation) to Python dictionary.
Args:
zeep_object (Any): The zeep object returned from ONVIF operations
Returns:
Python dictionary representation of the zeep object
Example:
```python linenums="1"
device = client.devicemgmt()
info = device.GetDeviceInformation()
info_dict = device.to_dict(info)
print(info_dict)
```
"""
try:
return (
{}
if zeep_object is None
else zeep.helpers.serialize_object(zeep_object)
)
except Exception as e: # pylint: disable=broad-except
logger.error("Failed to convert zeep object to dict: %s", e)
return {}