File size: 5,399 Bytes
b7d2408
 
d032bfc
 
b7d2408
 
03a45bc
b7d2408
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
d032bfc
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
b7d2408
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
03a45bc
b7d2408
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
"""CVAT API project methods."""

from __future__ import annotations

from typing import TYPE_CHECKING

from metrics_evaluation.schema.cvat import CvatApiLabelDefinition, CvatApiProjectDetails

from .retry import retry_with_backoff

# TYPE_CHECKING is False at runtime but True during static type checking.
# This allows us to import CvatApiClient for type hints without creating a circular
# import (client.py imports ProjectsMethods, and we need CvatApiClient for type hints).
# Benefits:
# - Avoids circular import errors at runtime
# - Provides proper type checking during development
# - No performance overhead (import only happens during type checking, not at runtime)
# This is a recommended pattern in modern Python (PEP 484, PEP 563).
# -- Claude Code
if TYPE_CHECKING:
    from .client import CvatApiClient


class ProjectsMethods:
    """Project-level operations for CVAT API."""

    def __init__(self, client: "CvatApiClient"):
        """Initialize project methods with client reference.

        Args:
            client: Parent CvatApiClient instance
        """
        self.client = client

    @retry_with_backoff(max_retries=3, initial_delay=1.0)
    def list(self, token: str | None = None) -> list[CvatApiProjectDetails]:
        """List all projects accessible to the user.

        Args:
            token: Authentication token (optional)

        Returns:
            List of project details objects
        """
        headers = self.client._get_headers(token)
        url = f"{self.client.cvat_host}/api/projects?page_size=1000"

        response = self.client._make_request(
            method="GET",
            url=url,
            headers=headers,
            resource_name="projects list",
        )

        response_data = response.json()
        return [
            CvatApiProjectDetails.model_validate(project)
            for project in response_data.get("results", [])
        ]

    @retry_with_backoff(max_retries=3, initial_delay=1.0)
    def get_project_details(
        self, project_id: int, token: str | None = None
    ) -> CvatApiProjectDetails:
        """Fetch project details.

        Args:
            project_id: The ID of the project
            token: Authentication token (optional)

        Returns:
            Project details object
        """
        headers = self.client._get_headers(token)
        url = f"{self.client.cvat_host}/api/projects/{project_id}"

        return self.client._make_request(
            method="GET",
            url=url,
            headers=headers,
            resource_name="project",
            resource_id=project_id,
            response_model=CvatApiProjectDetails,
        )

    @retry_with_backoff(max_retries=3, initial_delay=1.0)
    def get_project_labels(
        self, project_id: int, token: str | None = None
    ) -> list[CvatApiLabelDefinition]:
        """Fetch labels for a project.

        Args:
            project_id: The ID of the project
            token: Authentication token (optional)

        Returns:
            List of label definitions
        """
        headers = self.client._get_headers(token)
        url = f"{self.client.cvat_host}/api/labels?project_id={project_id}&page_size=1000"

        response = self.client._make_request(
            method="GET",
            url=url,
            headers=headers,
            resource_name="project labels",
            resource_id=project_id,
        )

        response_data = response.json()
        return [
            CvatApiLabelDefinition.model_validate(label)
            for label in response_data.get("results", [])
        ]

    @retry_with_backoff(max_retries=3, initial_delay=1.0)
    def get_project_job_ids(
        self, project_id: int, token: str | None = None
    ) -> list[int]:
        """Fetch all job IDs associated with a project.

        Args:
            project_id: The ID of the project
            token: Authentication token (optional)

        Returns:
            List of job IDs
        """
        from metrics_evaluation.schema.cvat import CvatApiJobsListResponse

        headers = self.client._get_headers(token)
        url = f"{self.client.cvat_host}/api/jobs?project_id={project_id}&page_size=1000&org={self.client.cvat_organization}"

        response = self.client._make_request(
            method="GET",
            url=url,
            headers=headers,
            resource_name="project jobs",
            resource_id=project_id,
            response_model=CvatApiJobsListResponse,
        )

        return [job.id for job in response.results]

    @retry_with_backoff(max_retries=3, initial_delay=1.0)
    def get_project_tasks(self, project_id: int, token: str | None = None) -> list[int]:
        """Fetch all task IDs for a project.

        Args:
            project_id: The ID of the project
            token: Authentication token (optional)

        Returns:
            List of task IDs
        """
        headers = self.client._get_headers(token)
        url = f"{self.client.cvat_host}/api/tasks?project_id={project_id}&page_size=1000"

        response = self.client._make_request(
            method="GET",
            url=url,
            headers=headers,
            resource_name="project tasks",
            resource_id=project_id,
        )

        response_data = response.json()
        return [task["id"] for task in response_data.get("results", [])]