[](https://github.com/cesbit/aiowmi/actions)
[](https://github.com/cesbit/aiowmi/releases)
# Python WMI
Windows Management Interface connector using asyncio for the Python language.
**Supports:**
- [x] NTLM Authentication
- [x] WMI Query (IWbemServices_ExecQuery)
- [x] Parsing of basic WMI Objects (int/float/str/datetime/array/references)
- [x] Optimized queries using the SmartEnum implementation
**Todo:**
- [ ] Kerberos Authentication
- [ ] Async WMI Query (IWbemServices_ExecQueryAsync)
- [ ] Other DCOM/RPC/WMI calls?
- [ ] Support for WMI Methods?
- [ ] Improve documentation
## Usage
The example below covers most of what is suppored by this library:
```python
import asyncio
import logging
import time
from aiowmi.connection import Connection
from aiowmi.query import Query
async def main():
host = '10.0.0.1' # ip address or hostname or fqdn
username = 'username'
password = 'password'
domain = '' # optional domain name
# Query has a default namespace 'root/cimv2'
queries = (
Query('SELECT * FROM Win32_OperatingSystem', namespace='root/cimv2'),
Query('SELECT * FROM Win32_NetworkAdapter'),
Query('SELECT * FROM Win32_LoggedOnUser'),
Query('SELECT * FROM Win32_PnpEntity'),
Query('SELECT Caption, Description, InstallDate, InstallDate2, '
'InstallLocation, InstallSource, InstallState, Language, '
'LocalPackage, Name, PackageCache, PackageCode, PackageName, '
'ProductID, RegCompany, RegOwner, SKUNumber, Transforms, '
'URLInfoAbout, URLUpdateInfo, Vendor, Version '
'FROM Win32_Product'),
Query('SELECT Name, DiskReadsPersec, DiskWritesPersec '
'FROM Win32_PerfFormattedData_PerfDisk_LogicalDisk'),
)
start = time.time()
conn = Connection(host, username, password, domain=domain)
service = None
await conn.connect()
try:
service = await conn.negotiate_ntlm()
for query in queries:
print(f"""
###############################################################################
# Start Query: {query.query}
###############################################################################
""")
async with query.context(conn, service) as qc:
async for props in qc.results():
# Function `get_properties(..)` accepts a few keyword
# arguments:
#
# ignore_defaults:
# Ignore default values. Set missing values to None
# if a value does not exist in the current class.
# ignore_defaults will always be True if
# ignore_missing is set to True.
# ignore_missing:
# If set to True, values missing in the current class
# will not be part of the result.
# load_qualifiers:
# Load the qualifiers of the properties. If False,
# the property qualifier_set will have the offset
# in the heap where the qualifiers are stored.
#
for name, prop in props.items():
print(name, '\n\t', prop.value)
if prop.is_reference():
# References can easy be queried using the
# get_reference(..) function. The function accepts
# a keyword argument `filter_props=[..]` with an
# optional list of properties to query. If omitted,
# the function returns all (*) properties.
res = await prop.get_reference(conn, service)
ref_props = res.get_properties(ignore_missing=True)
for name, prop in ref_props.items():
print('\t\t', name, '\n\t\t\t', prop.value)
print(f"""
----------------------------------- End Item ----------------------------------
""")
finally:
if service:
service.close()
conn.close()
end = time.time()
print('done in ', end-start)
if __name__ == '__main__':
logger = logging.getLogger()
logger.setLevel(logging.DEBUG)
ch = logging.StreamHandler()
ch.setLevel(logging.DEBUG)
formatter = logging.Formatter(
fmt='[%(levelname)1.1s %(asctime)s %(module)s:%(lineno)d] ' +
'%(message)s',
datefmt='%y%m%d %H:%M:%S',
style='%')
ch.setFormatter(formatter)
asyncio.run(main())
```
Raw data
{
"_id": null,
"home_page": "https://github.com/cesbit/aiowmi",
"name": "aiowmi",
"maintainer": "",
"docs_url": null,
"requires_python": "",
"maintainer_email": "",
"keywords": "WMI,Monitoring",
"author": "Jeroen van der Heijden",
"author_email": "jeroen@cesbit.com",
"download_url": "https://files.pythonhosted.org/packages/cc/29/0563375d9f73f1795a124e565bf24a921a836ed5cf79bd3cf5096e78b764/aiowmi-0.2.3.tar.gz",
"platform": null,
"description": "[](https://github.com/cesbit/aiowmi/actions)\n[](https://github.com/cesbit/aiowmi/releases)\n\n\n# Python WMI\n\nWindows Management Interface connector using asyncio for the Python language.\n\n**Supports:**\n- [x] NTLM Authentication\n- [x] WMI Query (IWbemServices_ExecQuery)\n- [x] Parsing of basic WMI Objects (int/float/str/datetime/array/references)\n- [x] Optimized queries using the SmartEnum implementation\n\n**Todo:**\n- [ ] Kerberos Authentication\n- [ ] Async WMI Query (IWbemServices_ExecQueryAsync)\n- [ ] Other DCOM/RPC/WMI calls?\n- [ ] Support for WMI Methods?\n- [ ] Improve documentation\n\n## Usage\n\nThe example below covers most of what is suppored by this library:\n\n```python\n\nimport asyncio\nimport logging\nimport time\nfrom aiowmi.connection import Connection\nfrom aiowmi.query import Query\n\n\nasync def main():\n\n host = '10.0.0.1' # ip address or hostname or fqdn\n username = 'username'\n password = 'password'\n domain = '' # optional domain name\n\n # Query has a default namespace 'root/cimv2'\n queries = (\n Query('SELECT * FROM Win32_OperatingSystem', namespace='root/cimv2'),\n Query('SELECT * FROM Win32_NetworkAdapter'),\n Query('SELECT * FROM Win32_LoggedOnUser'),\n Query('SELECT * FROM Win32_PnpEntity'),\n Query('SELECT Caption, Description, InstallDate, InstallDate2, '\n 'InstallLocation, InstallSource, InstallState, Language, '\n 'LocalPackage, Name, PackageCache, PackageCode, PackageName, '\n 'ProductID, RegCompany, RegOwner, SKUNumber, Transforms, '\n 'URLInfoAbout, URLUpdateInfo, Vendor, Version '\n 'FROM Win32_Product'),\n Query('SELECT Name, DiskReadsPersec, DiskWritesPersec '\n 'FROM Win32_PerfFormattedData_PerfDisk_LogicalDisk'),\n )\n\n start = time.time()\n\n conn = Connection(host, username, password, domain=domain)\n service = None\n await conn.connect()\n try:\n service = await conn.negotiate_ntlm()\n\n for query in queries:\n print(f\"\"\"\n###############################################################################\n# Start Query: {query.query}\n###############################################################################\n\"\"\")\n async with query.context(conn, service) as qc:\n async for props in qc.results():\n # Function `get_properties(..)` accepts a few keyword\n # arguments:\n #\n # ignore_defaults:\n # Ignore default values. Set missing values to None\n # if a value does not exist in the current class.\n # ignore_defaults will always be True if\n # ignore_missing is set to True.\n # ignore_missing:\n # If set to True, values missing in the current class\n # will not be part of the result.\n # load_qualifiers:\n # Load the qualifiers of the properties. If False,\n # the property qualifier_set will have the offset\n # in the heap where the qualifiers are stored.\n #\n for name, prop in props.items():\n print(name, '\\n\\t', prop.value)\n\n if prop.is_reference():\n # References can easy be queried using the\n # get_reference(..) function. The function accepts\n # a keyword argument `filter_props=[..]` with an\n # optional list of properties to query. If omitted,\n # the function returns all (*) properties.\n res = await prop.get_reference(conn, service)\n ref_props = res.get_properties(ignore_missing=True)\n for name, prop in ref_props.items():\n print('\\t\\t', name, '\\n\\t\\t\\t', prop.value)\n\n print(f\"\"\"\n----------------------------------- End Item ----------------------------------\n\"\"\")\n finally:\n if service:\n service.close()\n conn.close()\n end = time.time()\n print('done in ', end-start)\n\nif __name__ == '__main__':\n logger = logging.getLogger()\n logger.setLevel(logging.DEBUG)\n\n ch = logging.StreamHandler()\n ch.setLevel(logging.DEBUG)\n\n formatter = logging.Formatter(\n fmt='[%(levelname)1.1s %(asctime)s %(module)s:%(lineno)d] ' +\n '%(message)s',\n datefmt='%y%m%d %H:%M:%S',\n style='%')\n\n ch.setFormatter(formatter)\n\n asyncio.run(main())\n\n\n```\n\n\n",
"bugtrack_url": null,
"license": "",
"summary": "Python WMI Queries",
"version": "0.2.3",
"split_keywords": [
"wmi",
"monitoring"
],
"urls": [
{
"comment_text": "",
"digests": {
"md5": "62d536c2b4bb2351d92c70683e2cd12c",
"sha256": "37ffa6baf0a13257d00a195c5f8e77d9aa55e8ed9dd0dcb5ec488d593e0b8010"
},
"downloads": -1,
"filename": "aiowmi-0.2.3.tar.gz",
"has_sig": false,
"md5_digest": "62d536c2b4bb2351d92c70683e2cd12c",
"packagetype": "sdist",
"python_version": "source",
"requires_python": null,
"size": 76464,
"upload_time": "2022-12-12T15:02:48",
"upload_time_iso_8601": "2022-12-12T15:02:48.441176Z",
"url": "https://files.pythonhosted.org/packages/cc/29/0563375d9f73f1795a124e565bf24a921a836ed5cf79bd3cf5096e78b764/aiowmi-0.2.3.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2022-12-12 15:02:48",
"github": true,
"gitlab": false,
"bitbucket": false,
"github_user": "cesbit",
"github_project": "aiowmi",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"requirements": [
{
"name": "pycryptodome",
"specs": [
[
">=",
"3.14.0"
]
]
}
],
"lcname": "aiowmi"
}
JSON
