client.py - laba3 - qpfilw/restapi2

client.py
224 строки7 KB
qpfilw
ready client
15 дек 2025, 18:20
15 дек 2025, 18:2097d8d2b
import requests
import json
import uuid
import time

BASE_URL = "http://localhost:5086"

def print_response(response):
    print(f"Status Code: {response.status_code}")
    try:
        print("Response Body:", json.dumps(response.json(), indent=4))
    except:
        print("Response Body:", response.text)
    print("\n")

# Authentication
def authenticate():
    url = f"{BASE_URL}/api/auth/token"
    payload = {
        "Username": "admin",
        "Password": "admin"
    }
    response = requests.post(url, json=payload)
    print("Authentication:")
    print_response(response)
    if response.status_code == 200:
        token = response.json()["access_Token"]
        return token
    else:
        raise Exception("Authentication failed")

# Create a project
def create_project(token, name):
    url = f"{BASE_URL}/api/v1/projects"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    payload = {
        "Name": name
    }
    response = requests.post(url, headers=headers, json=payload)
    print(f"Create Project '{name}':")
    print_response(response)
    if response.status_code == 201:
        return response.json()["id"]
    return None

# Get all projects
def get_all_projects():
    url = f"{BASE_URL}/api/v1/projects"
    response = requests.get(url)
    print("Get All Projects:")
    print_response(response)

# Get project by ID
def get_project_by_id(project_id):
    url = f"{BASE_URL}/api/v1/projects/{project_id}"
    response = requests.get(url)
    print(f"Get Project by ID {project_id}:")
    print_response(response)

# Update project
def update_project(token, project_id, new_name):
    url = f"{BASE_URL}/api/v1/projects/{project_id}"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    payload = {
        "Name": new_name
    }
    response = requests.put(url, headers=headers, json=payload)
    print(f"Update Project {project_id} to '{new_name}':")
    print_response(response)

# Delete project
def delete_project(token, project_id):
    url = f"{BASE_URL}/api/v1/projects/{project_id}"
    headers = {
        "Authorization": f"Bearer {token}"
    }
    response = requests.delete(url, headers=headers)
    print(f"Delete Project {project_id}:")
    print_response(response)

# Create task
def create_task(token, project_id, title, is_done=False, version="1.0", priority=None, idempotency_key=None):
    url = f"{BASE_URL}/api/v{version}/tasks"
    if idempotency_key is None:
        idempotency_key = str(uuid.uuid4())
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
        "Idempotency-Key": idempotency_key
    }
    if version == "1.0":
        payload = {
            "Title": title,
            "IsDone": is_done,
            "ProjectId": project_id
        }
    else:  # v2.0
        payload = {
            "Title": title,
            "IsDone": is_done,
            "ProjectId": project_id,
            "Priority": priority
        }
    response = requests.post(url, headers=headers, json=payload)
    print(f"Create Task '{title}' in Project {project_id} (v{version}):")
    print_response(response)
    if response.status_code == 200:
        return response.json()["id"]
    return None

# idempotency with same key
def demonstrate_idempotency(token, project_id, title, idempotency_key):
    url = f"{BASE_URL}/api/v1.0/tasks"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
        "Idempotency-Key": idempotency_key
    }
    payload = {
        "Title": title,
        "IsDone": False,
        "ProjectId": project_id
    }
    response = requests.post(url, headers=headers, json=payload)
    print(f"Repeat Create Task with same Idempotency-Key '{idempotency_key}':")
    print_response(response)

# Get all tasks with pagination
def get_all_tasks(token, page=1, page_size=10):
    url = f"{BASE_URL}/api/v1/tasks?page={page}&pageSize={page_size}"
    headers = {
        "Authorization": f"Bearer {token}"
    }
    response = requests.get(url, headers=headers)
    print(f"Get All Tasks (Page {page}, Size {page_size}):")
    print_response(response)

# Get task by ID
def get_task_by_id(token, task_id, include=None):
    url = f"{BASE_URL}/api/v1/tasks/{task_id}"
    if include:
        url += f"?include={include}"
    headers = {
        "Authorization": f"Bearer {token}"
    }
    response = requests.get(url, headers=headers)
    print(f"Get Task by ID {task_id} (include={include}):")
    print_response(response)

# Update task
def update_task(token, task_id, new_title, is_done=True):
    url = f"{BASE_URL}/api/v1/tasks/{task_id}"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    payload = {
        "Title": new_title,
        "IsDone": is_done
    }
    response = requests.put(url, headers=headers, json=payload)
    print(f"Update Task {task_id} to '{new_title}':")
    print_response(response)

# Delete task
def delete_task(token, task_id):
    url = f"{BASE_URL}/api/v1/tasks/{task_id}"
    headers = {
        "Authorization": f"Bearer {token}"
    }
    response = requests.delete(url, headers=headers)
    print(f"Delete Task {task_id}:")
    print_response(response)

def demonstrate_rate_limiting(token):
    url = f"{BASE_URL}/api/v1/projects"
    headers = {
        "Authorization": f"Bearer {token}"
    }
    for i in range(25):
        response = requests.get(url, headers=headers)
        print(f"Rate Limit Test Request {i+1}: Status {response.status_code}")
        if response.status_code == 429:
            print("Rate limit hit!")
            retry_after = response.headers.get("Retry-After", "unknown")
            print(f"Retry after: {retry_after} seconds")
            break
        time.sleep(2)

if __name__ == "__main__":
    try:
        token = authenticate()

        project_id = create_project(token, "Test Project")
        if project_id:
            get_all_projects()
            get_project_by_id(project_id)
            update_project(token, project_id, "Updated Project")

        if project_id:
            idempotency_key = str(uuid.uuid4())
            task_id = create_task(token, project_id, "Test Task", version="1.0", idempotency_key=idempotency_key)
            if task_id:
                demonstrate_idempotency(token, project_id, "Test Task", idempotency_key)
                
                get_all_tasks(token)
                get_task_by_id(token, task_id, include="project")
                update_task(token, task_id, "Updated Task")
                delete_task(token, task_id)

        if project_id:
            create_task(token, project_id, "Priority Task", priority=5, version="2.0")

        if project_id:
            delete_project(token, project_id)

    except Exception as e:
        print(f"Error: {e}")