需要帮助将 API PUT 方法添加到 Python 脚本

Question

我正在使用以下脚本从服务器收集库存信息并将其发送到名为 Device42 的产品。该脚本当前有效，但是我尝试添加的其中一个 API 使用 PUT 而不是 POST。我不是程序员，只是开始在这个脚本中使用 python。该脚本使用的是铁蟒。这个脚本可以使用PUT方法吗？

"""
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
"""
##################################################
# a sample script to show how to use
#   /api/ip/add-or-update
#   /api/device/add-or-update
#
# requires ironPython (http://ironpython.codeplex.com/) and
# powershell (http://support.microsoft.com/kb/968929)
##################################################

import clr

clr.AddReference('System.Management.Automation')

from System.Management.Automation import (
PSMethod, RunspaceInvoke
)
RUNSPACE = RunspaceInvoke()

import urllib
import urllib2
import traceback
import base64
import math
import ssl
import functools
BASE_URL='https://device42_URL'

API_DEVICE_URL=BASE_URL+'/api/1.0/devices/'
API_IP_URL    =BASE_URL+'/api/1.0/ips/'
API_PART_URL=BASE_URL+'/api/1.0/parts/'
API_MOUNTPOINT_URL=BASE_URL+'/api/1.0/device/mountpoints/'
API_CUSTOMFIELD_URL=BASE_URL+'/api/1.0/device/custom_field/'

USER     ='usernme'
PASSWORD ='password'

old_init = ssl.SSLSocket.__init__
@functools.wraps(old_init)
def init_with_tls1(self, *args, **kwargs):
kwargs['ssl_version'] = ssl.PROTOCOL_TLSv1
old_init(self, *args, **kwargs)
ssl.SSLSocket.__init__ = init_with_tls1

def post(url, params):
"""
http post with basic-auth
params is dict like object
"""
try:
    data= urllib.urlencode(params) # convert to ascii chars
    headers = {
        'Authorization' : 'Basic '+ base64.b64encode(USER + ':' + PASSWORD),
        'Content-Type'  : 'application/x-www-form-urlencoded'
    }

    req = urllib2.Request(url, data, headers)

    print '---REQUEST---',req.get_full_url()
    print req.headers
    print req.data

    reponse = urllib2.urlopen(req)

    print '---RESPONSE---'
    print reponse.getcode()
    print reponse.info()
    print reponse.read()
except urllib2.HTTPError as err:
    print '---RESPONSE---'
    print err.getcode()
    print err.info()
    print err.read()
except urllib2.URLError as err:
    print '---RESPONSE---'
    print err

def to_ascii(s):
    # ignore non-ascii chars
    return s.encode('ascii','ignore')

def wmi(query):
    return [dict([(prop.Name, prop.Value) for prop in psobj.Properties]) for psobj in RUNSPACE.Invoke(query)]
def closest_memory_assumption(v):
    return int(256 * math.ceil(v / 256.0))

def add_or_update_device():
    computer_system  = wmi('Get-WmiObject Win32_ComputerSystem -Namespace "root\CIMV2"')[0] # take first
    bios             = wmi('Get-WmiObject Win32_BIOS -Namespace "root\CIMV2"')[0]
    operating_system = wmi('Get-WmiObject Win32_OperatingSystem -Namespace "root\CIMV2"')[0]
    environment      = wmi('Get-WmiObject Win32Reg_ESFFarmNode -Namespace "root\CIMV2"')[0]
    mem              = closest_memory_assumption(int(computer_system.get('TotalPhysicalMemory')) / 1047552)
    dev_name         = to_ascii(computer_system.get('Name')).upper()
    fqdn_name        = to_ascii(computer_system.get('Name')).upper() + '.' + to_ascii(computer_system.get('Domain')).lower()
device = {
    'memory'        : mem,
    'os'            : to_ascii(operating_system.get('Caption')),
    'osver'         : operating_system.get('OSArchitecture'),
    'osmanufacturer': to_ascii(operating_system.get('Manufacturer')),
    'osserial'      : operating_system.get('SerialNumber'),
    'osverno'       : operating_system.get('Version'),
    'service_level' : environment.get('Environment'),
    'notes'         : 'Test w/ Change to Device name collection'
}
devicedmn = ''
for dmn in ['Domain1', 'Domain2', 'Domain3', 'Domain4', 'Domain5']:
    if dmn == to_ascii(computer_system.get('Domain')).strip():
        devicedmn = 'Domain'
        device.update({ 'name' : fqdn_name, })
        break    
if devicedmn != 'Domain':
    device.update({
        'name': dev_name,
        })
manufacturer = ''
for mftr in ['VMware, Inc.', 'Bochs', 'KVM', 'QEMU', 'Microsoft Corporation', 'Xen']:
    if mftr == to_ascii(computer_system.get('Manufacturer')).strip():
        manufacturer = 'virtual'
        device.update({ 'manufacturer' : 'vmware', })
        break    
if manufacturer != 'virtual':
    device.update({
        'manufacturer': to_ascii(computer_system.get('Manufacturer')).strip(),
        'hardware': to_ascii(computer_system.get('Model')).strip(),
        'serial_no': to_ascii(bios.get('SerialNumber')).strip(),
        'type': 'Physical',
        })    
cpucount = 0
for cpu in wmi('Get-WmiObject Win32_Processor  -Namespace "root\CIMV2"'):
    cpucount += 1
    cpuspeed = cpu.get('MaxClockSpeed')
    cpucores = cpu.get('NumberOfCores')
if cpucount > 0:

    device.update({
        'cpucount': cpucount,
        'cpupower': cpuspeed,
        'cpucore':  cpucores,
        })
hddcount = 0
hddsize = 0
for hdd in wmi('Get-WmiObject Win32_LogicalDisk  -Namespace "root\CIMV2" | where{$_.Size -gt 1}'):
    hddcount += 1
    hddsize += hdd.get('Size') / 1073741742
if hddcount > 0:

    device.update({
        'hddcount': hddcount,
        'hddsize': hddsize,
        })
post(API_DEVICE_URL, device)

for hdd in wmi('Get-WmiObject Win32_LogicalDisk  -Namespace "root\CIMV2" | where{$_.Size -gt 1}'):
    mountpoint = {
          'mountpoint' : hdd.get('Name'),
          'label' : hdd.get('Caption'),
          'fstype' : hdd.get('FileSystem'),
          'capacity' : hdd.get('Size') / 1024 / 1024,
          'free_capacity' : hdd.get('FreeSpace') / 1024 / 1024,
          'device' : dev_name,
          'assignment' : 'Device',
    }
    post(API_MOUNTPOINT_URL, mountpoint)

network_adapter_configuration = wmi('Get-WmiObject Win32_NetworkAdapterConfiguration -Namespace "root\CIMV2" | where{$_.IPEnabled -eq "True"}')

for ntwk in network_adapter_configuration:
    for ipaddr in ntwk.get('IPAddress'):
        ip = {
            'ipaddress'  : ipaddr,
            'macaddress' : ntwk.get('MACAddress'),
            'label'      : ntwk.get('Description'),
            'device'     : dev_name,
        }
        post(API_IP_URL, ip)

def main():
    try:
        add_or_update_device()
    except:
        traceback.print_exc()

if __name__ == "__main__":
    main()

Answer 1

首先，您需要了解 PUT 和 POST 之间的区别。我会把它写出来，但是社区的另一位成员对这两个 here 给出了很好的描述。

现在，是的，您可以在该脚本中使用请求。下面是一个通过 python 使用 requests 库的例子，如果你安装了 pip 来安装请求，像这样安装它：

pip install requests

现在，为了避免使用 Requests 库的一些示例，可以在 here 找到文档。

HTTP 获取请求。所以对于这个例子，你从请求库中调用 get 函数，将 url 作为参数，然后你可以从返回的 touple 中打印出文本。由于 GET 会返回一些东西，它通常会在 touple 的文本部分允许您打印它。

r = requests.get('http://urlhere.com/apistuffhere')
print(r.text)

HTTP POST：发布到 url，取决于 API 的设置方式，它会返回一些东西，它通常用于错误处理，但您还必须传入参数。这是对新用户条目的 POST 请求的示例。同样，您可以打印来自 touple 的文本以检查来自 API 的响应

payload = {'username': 'myloginname', 'password': 'passwordhere'}
r = requests.post('https://testlogin.com/newuserEntry', params=payload)
print(r.text)

或者你可以只打印 r 并且它应该返回一个响应 200 这应该是成功的。

对于 PUT：您必须记住 put 响应不能被缓存，因此您可以将数据发布到 PUT url，但您不会知道是否有错误，但使用与 POST 相同的语法。我没有尝试使用 Request 库在 PUT 请求中打印文本响应，因为我没有在我编写的任何 API 中使用 PUT。

requests.put('http://urlhere.com/putextension')

现在要在您的代码中实现这一点，您已经有了 url 的基础，在您的登录帖子中只需执行以下操作：

payload = {'username': USERNAME, 'passwd':PASSWORD}
r = requests.post('https://loginurlhere.com/', params=payload)
#check response by printing text
print (r.text)

至于将数据放入您的 api 扩展，让我们假设您已经准备好包含所需信息的有效负载变量，例如 API 设备扩展：

requests.put(API_DEVICE, params=payload)

这应该放到网址中。如果您有任何问题，请在下面发表评论，如果您愿意，我可以回答。

Answer 2

我的标准答案是将 urllib2 替换为 Requests 包。它使 HTTP 工作变得更加容易。

但请查看 this SO answer 以获取“破解”以使 PUT 正常工作。