users.py 3.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123
  1. #!/usr/bin/env python3
  2. import requests
  3. import argparse
  4. from datetime import datetime, timedelta
  5. def view(
  6. url,
  7. token,
  8. name=None,
  9. group_name=None,
  10. ):
  11. headers = {"Authorization": f"Bearer {token}"}
  12. pageSize = 30
  13. params = {
  14. "name": name,
  15. "group_name": group_name,
  16. }
  17. params = {
  18. k: "%" + v + "%" if (v != "-" and "%" not in v) else v
  19. for k, v in params.items()
  20. if v is not None
  21. }
  22. params["pageSize"] = pageSize
  23. users = []
  24. current = 1
  25. while True:
  26. params["current"] = current
  27. response = requests.get(f"{url}/api/users", headers=headers, params=params)
  28. response_json = response.json()
  29. data = response_json.get("data", [])
  30. users.extend(data)
  31. total = response_json.get("total", 0)
  32. current += pageSize
  33. if len(data) < pageSize or current > total:
  34. break
  35. return users
  36. def check(response):
  37. if response.status_code == 200:
  38. try:
  39. response_json = response.json()
  40. return response_json
  41. except ValueError:
  42. return response.text or "Success"
  43. else:
  44. return "Failed", response.status_code, response.text
  45. def disable(url, token, guid, name):
  46. print("Disable", name)
  47. headers = {"Authorization": f"Bearer {token}"}
  48. response = requests.post(f"{url}/api/users/{guid}/disable", headers=headers)
  49. return check(response)
  50. def enable(url, token, guid, name):
  51. print("Enable", name)
  52. headers = {"Authorization": f"Bearer {token}"}
  53. response = requests.post(f"{url}/api/users/{guid}/enable", headers=headers)
  54. return check(response)
  55. def delete(url, token, guid, name):
  56. print("Delete", name)
  57. headers = {"Authorization": f"Bearer {token}"}
  58. response = requests.delete(f"{url}/api/users/{guid}", headers=headers)
  59. return check(response)
  60. def main():
  61. parser = argparse.ArgumentParser(description="User manager")
  62. parser.add_argument(
  63. "command",
  64. choices=["view", "disable", "enable", "delete"],
  65. help="Command to execute",
  66. )
  67. parser.add_argument("--url", required=True, help="URL of the API")
  68. parser.add_argument(
  69. "--token", required=True, help="Bearer token for authentication"
  70. )
  71. parser.add_argument("--name", help="User name")
  72. parser.add_argument("--group_name", help="Group name")
  73. args = parser.parse_args()
  74. while args.url.endswith("/"): args.url = args.url[:-1]
  75. users = view(
  76. args.url,
  77. args.token,
  78. args.name,
  79. args.group_name,
  80. )
  81. if args.command == "view":
  82. for user in users:
  83. print(user)
  84. elif args.command == "disable":
  85. for user in users:
  86. response = disable(args.url, args.token, user["guid"], user["name"])
  87. print(response)
  88. elif args.command == "enable":
  89. for user in users:
  90. response = enable(args.url, args.token, user["guid"], user["name"])
  91. print(response)
  92. elif args.command == "delete":
  93. for user in users:
  94. response = delete(args.url, args.token, user["guid"], user["name"])
  95. print(response)
  96. if __name__ == "__main__":
  97. main()