Compare commits

...

6 Commits

Author SHA1 Message Date
8be5b5f1c8 cli to view workouts
Some checks failed
Nix Tests / nix-test (nix-runner) (push) Failing after 3m35s
Python Tests / python-test (push) Failing after 52s
2025-11-01 23:38:19 -05:00
2dbd4a80a1 adding upload testing, etc.
All checks were successful
Nix Tests / nix-test (nix-runner) (push) Successful in 7m39s
Python Tests / python-test (push) Successful in 8m12s
2025-10-31 17:58:01 -05:00
64327c73e9 fmt 2025-10-26 18:22:21 -05:00
f847a0273f hats 2025-10-26 18:20:46 -05:00
f516e4f8e5 move again 2025-10-23 12:07:59 -05:00
c8c7680fac rename 2025-10-23 12:07:09 -05:00
29 changed files with 1962 additions and 283 deletions

1
.gitignore vendored
View File

@@ -157,3 +157,4 @@ result
# for local dev might dump cred stuff here # for local dev might dump cred stuff here
cred/ cred/
local/

130
CLAUDE.md
View File

@@ -4,7 +4,7 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co
## Project Overview ## Project Overview
This is a Python CLI client for the trygo webapp built with a hybrid development environment using both uv and Nix. The project has dual entry points: `hello` (hello_world package) and `taco` (trygo_py_cliclient.cli package). This is a Python CLI client for the TryGo webapp built with a hybrid development environment using both uv and Nix. The project provides a complete CLI interface for the TryGo Activity Files API, allowing users to upload, manage, and interact with activity files (.fit files). The project has dual entry points: `hello` (hello_world package) and `taiga` (taiga_pycli.cli package).
## Development Commands ## Development Commands
@@ -21,19 +21,107 @@ This is a Python CLI client for the trygo webapp built with a hybrid development
- `uv run mypy src` - Type checking - `uv run mypy src` - Type checking
- `uv run pytest --snapshot-update` - Update test snapshots - `uv run pytest --snapshot-update` - Update test snapshots
## CLI Commands
The `taiga` CLI provides the following commands:
### Authentication Commands
- `taiga register --email <email> --password <password> --display-name <name>` - Register new user account
- `taiga login --email <email> --password <password>` - Authenticate and store JWT token
### Activity File Commands
- `taiga activities upload <file_path>` - Upload activity files (.fit files) to the server
- Requires authentication (login first)
- Validates file existence and readability
- Uses current timestamp automatically
- Returns activity file metadata on success
- `taiga activities ls` - List all activity files for the authenticated user
- Shows ID, timestamp, and creation date in tabular format
- Requires authentication (login first)
- `taiga activities download <id> [--output <path>]` - Download activity file by ID
- Downloads activity file to specified path or defaults to activity_{id}.fit
- Requires authentication (login first)
### Other Commands
- `taiga hat` - Hat management commands (development/testing feature)
### Examples
```bash
# Register and login
taiga register --email "user@example.com" --password "mypassword" --display-name "My Name"
taiga login --email "user@example.com" --password "mypassword"
# Activity file management
taiga activities upload activity.fit
taiga activities upload local/test-data/test.fit
taiga activities ls
taiga activities download 1
taiga activities download 1 --output my_activity.fit
```
## Architecture ## Architecture
### Package Structure ### Package Structure
- `src/hello_world/` - Hello world demonstration package - `src/hello_world/` - Hello world demonstration package
- `src/trygo_py_cliclient/` - Main CLI client package - `src/taiga_pycli/` - Main CLI client package
- `cli/` - Command-line interface and argument parsing - `cli/` - Command-line interface and argument parsing
- `config/` - Configuration management with TOML support - `config/` - Configuration management with TOML support
- `service/` - Backend API service layer
- `models.py` - Data models for API requests/responses
## Upload Functionality
The upload feature provides seamless integration with the TryGo Activity Files API:
### Implementation Details
- **ActivityFile Model**: Dataclass representing activity file metadata with fields for id, timestamp, file_repo_hash, created_at, updated_at, and user_id
- **BackendService.upload_activity_file()**: Handles multipart file uploads with proper JWT authentication
- **File Validation**: Checks file existence, readability, and path validity before upload
- **Error Handling**: Comprehensive error handling for authentication, validation, network, and server errors
- **Automatic Timestamping**: Uses current time for activity timestamp (no custom timestamp needed)
### Supported File Formats
- Primary: `.fit` files (Garmin activity files)
- The API accepts any file format, but the CLI is designed for activity files
### Authentication Requirements
- Must be logged in with valid JWT token before uploading
- Token is automatically stored in `cred/token` after login
- Token is automatically included in upload requests
### Error Scenarios Handled
- File not found or not readable
- Not authenticated (missing/invalid token)
- Network connectivity issues
- Server errors (400, 500, etc.)
- Invalid API responses
### Configuration System ### Configuration System
- Uses dataclasses for type-safe configuration (`src/trygo_py_cliclient/config/config.py`) - Uses dataclasses for type-safe configuration (`src/taiga_pycli/config/config.py`)
- TOML-based configuration files (`config.toml`) - TOML-based configuration files (`config.toml`)
- Supports logging configuration and general application settings - Supports logging configuration and general application settings
### TryGo Activity Files API Integration
- **BackendService Class**: Centralized API client with session management
- JWT token authentication with automatic header injection
- Token persistence in `cred/token` file
- Comprehensive error handling with custom exception types
- Support for authentication endpoints (`/auth/register`, `/auth/tokens`)
- Support for activity file endpoints (`/activity_files`, `/activity_files/{id}/download`)
- Methods: `upload_activity_file()`, `list_activity_files()`, `download_activity_file()`
- **Data Models**: Type-safe dataclasses for API interactions
- `ActivityFile`: Response model for activity file metadata
- `RegisterRequest`/`LoginRequest`: Authentication request models
- `AuthResponse`: Authentication response model
- **Error Handling Strategy**: Custom exception hierarchy
- `AuthenticationError`: 401 responses and missing tokens
- `ValidationError`: 400 responses and file validation issues
- `NetworkError`: Connection timeouts and network issues
- `ServerError`: 5xx server responses
- `TryGoAPIError`: General API errors
### Development Environment ### Development Environment
The project supports both uv and Nix environments: The project supports both uv and Nix environments:
- **uv mode**: Standard Python toolchain with uv for dependency management - **uv mode**: Standard Python toolchain with uv for dependency management
@@ -44,6 +132,7 @@ The project supports both uv and Nix environments:
- pytest with syrupy for snapshot testing - pytest with syrupy for snapshot testing
- Coverage reporting enabled (50% minimum) - Coverage reporting enabled (50% minimum)
- Test configuration in pyproject.toml with XML and HTML output - Test configuration in pyproject.toml with XML and HTML output
- Custom testing scripts for upload functionality validation
### Code Quality ### Code Quality
- ruff for linting and formatting (tab indentation style) - ruff for linting and formatting (tab indentation style)
@@ -51,9 +140,42 @@ The project supports both uv and Nix environments:
- flake8 for additional linting - flake8 for additional linting
- treefmt.nix for comprehensive formatting in Nix environment - treefmt.nix for comprehensive formatting in Nix environment
## Testing and Development Scripts
### Development Scripts
- `scripts/simple_create_user.sh` - Quick user registration script for testing
- Creates test user with email "test@example.com" and password "test"
- Used for development and testing workflows
- `scripts/test_upload.sh` - Comprehensive upload functionality testing
- Tests authentication requirements (upload without login should fail)
- Tests successful login workflow
- Tests successful file upload
- Tests error handling (non-existent files)
- Provides complete end-to-end validation
### Test Data
- `local/test-data/` - Directory containing real .fit files for testing
- `test.fit`, `test2.fit`, `test3.fit` - Sample activity files
- Organized separately from development files
- Used by test scripts for realistic upload testing
### Testing Workflow
```bash
# Run complete upload test suite
./scripts/test_upload.sh
# Manual testing steps
./scripts/simple_create_user.sh # Create test user (if needed)
taiga login --email "test@example.com" --password "test"
taiga activities upload local/test-data/test.fit
taiga activities ls
taiga activities download 1
```
## Entry Points ## Entry Points
- `hello` command maps to `hello_world:main` - `hello` command maps to `hello_world:main`
- `taco` command maps to `trygo_py_cliclient.cli:main` - `taiga` command maps to `taiga_pycli.cli:main`
## Configuration Files ## Configuration Files
- `pyproject.toml` - Python project configuration and dependencies - `pyproject.toml` - Python project configuration and dependencies

View File

@@ -2,3 +2,7 @@
backend_base_url = "http://localhost:8080" backend_base_url = "http://localhost:8080"
backend_user = "test@example.com" backend_user = "test@example.com"
backend_pw = "test" backend_pw = "test"
log_file = "local/logs/taiga_cli.log"
log_stream = false

View File

@@ -8,11 +8,12 @@ dependencies = [
"requests>=2.31.0", "requests>=2.31.0",
"dacite>=1.9.2", "dacite>=1.9.2",
"tomli>=2.2.1", "tomli>=2.2.1",
"types-requests>=2.32.4.20250913",
] ]
[project.scripts] [project.scripts]
hello = "hello_world:main" hello = "hello_world:main"
taco = "trygo_py_cliclient.cli:main" taiga = "taiga_pycli.cli:main"
[build-system] [build-system]
requires = ["hatchling"] requires = ["hatchling"]
@@ -41,11 +42,21 @@ sources = ["src"]
[tool.pytest.ini_options] [tool.pytest.ini_options]
testpaths = ["tests"] testpaths = ["tests"]
# Uncomment to care about coverage # Uncomment to care about coverage
addopts = "--junitxml pytest.xml --cov src --cov-report=xml:coverage.xml --cov-fail-under=50 --cov-report=html" addopts = "--junitxml pytest.xml --cov src --cov-report=xml:coverage.xml --cov-fail-under=1 --cov-report=html"
junit_family = "xunit1" junit_family = "xunit1"
log_format = "%(asctime)s | %(levelname)s | %(pathname)s:%(lineno)d | %(message)s" log_format = "%(asctime)s | %(levelname)s | %(pathname)s:%(lineno)d | %(message)s"
log_level = "WARNING" log_level = "WARNING"
[tool.pyright]
executionEnvironments = [
{ root = "src" }
]
exclude = [
".venv"
]
venvPath = "."
venv = ".venv"
# [tool.mypy] # [tool.mypy]
# If you need this # If you need this
# plugins = "numpy.typing.mypy_plugin" # plugins = "numpy.typing.mypy_plugin"

View File

@@ -1,5 +1,5 @@
#!/usr/bin/env bash #!/usr/bin/env bash
set -Eeuox pipefail set -Eeuo pipefail
banner() { banner() {
echo "========================================================" echo "========================================================"
@@ -9,4 +9,20 @@ banner() {
# utility script for easy testing # utility script for easy testing
uv run taco register --display-name "Display Test" --email "test@example.com" --password "test" banner "Checking if test user exists"
if uv run taiga login --email "test@example.com" --password "test" 2>/dev/null; then
echo "Test user already exists, skipping registration"
else
banner "Creating test user"
uv run taiga register --display-name "Display Test" --email "test@example.com" --password "test"
banner "Logging in test user"
uv run taiga login --email "test@example.com" --password "test"
fi
banner "Uploading test files"
uv run taiga workouts upload local/test-data/test.fit
uv run taiga workouts upload local/test-data/test2.fit
uv run taiga workouts upload local/test-data/test3.fit
banner "Setup complete!"

46
scripts/test_upload.sh Normal file
View File

@@ -0,0 +1,46 @@
#!/usr/bin/env bash
set -Eeuox pipefail
banner() {
echo "========================================================"
echo " $*"
echo "========================================================"
}
# Test script for activities functionality
banner "Testing Activities Functionality"
# Check if we have test files
if [ ! -d "local/test-data" ] || [ -z "$(ls -A local/test-data/*.fit 2>/dev/null)" ]; then
echo "Error: No test .fit files found in local/test-data/"
echo "Please ensure test files are available."
exit 1
fi
# Test 1: Try upload without authentication (should fail)
banner "Test 1: Upload without authentication (should fail)"
echo "This should show an authentication error:"
uv run taiga activities upload local/test-data/test.fit || echo "Expected failure - not authenticated"
# Test 2: Login first
banner "Test 2: Login with test user"
uv run taiga login --email "test@example.com" --password "test"
# Test 3: Upload a file (should succeed)
banner "Test 3: Upload test file (should succeed)"
uv run taiga activities upload local/test-data/test.fit
# Test 4: List activities (should show uploaded file)
banner "Test 4: List activities (should show uploaded file)"
uv run taiga activities ls
# Test 5: Try uploading non-existent file (should fail)
banner "Test 5: Upload non-existent file (should fail)"
uv run taiga activities upload nonexistent.fit || echo "Expected failure - file not found"
# Test 6: Try download (note: this may fail if server doesn't support download endpoint yet)
banner "Test 6: Download test (may fail if endpoint not implemented)"
uv run taiga activities download 1 --output downloaded_activity.fit || echo "Expected failure - download endpoint may not be implemented yet"
banner "Activities testing complete!"

View File

@@ -0,0 +1,48 @@
import argparse
import pathlib
import logging
import taiga_pycli.config
import taiga_pycli.cli.common
import taiga_pycli.cli.register
import taiga_pycli.cli.login
import taiga_pycli.cli.hats
import taiga_pycli.cli.activities
import taiga_pycli.cli.workouts
_logger = logging.getLogger(__name__)
def parse_args():
parser = argparse.ArgumentParser(
"taiga_pycli", formatter_class=argparse.ArgumentDefaultsHelpFormatter
)
parser.add_argument(
"--config-file", type=str, help="config file location", default="config.toml"
)
subparsers = parser.add_subparsers(dest="cmd", required=True)
taiga_pycli.cli.register.setup_parser(subparsers)
taiga_pycli.cli.login.setup_parser(subparsers)
taiga_pycli.cli.hats.setup_parser(subparsers)
taiga_pycli.cli.activities.setup_parser(subparsers)
taiga_pycli.cli.workouts.setup_parser(subparsers)
args = parser.parse_args()
return args
def main():
args = parse_args()
config = taiga_pycli.config.read_config(pathlib.Path(args.config_file))
taiga_pycli.cli.common.set_up_logging(
config,
)
_logger.info(f"Got args {args=}")
_logger.info(f"Loaded config {config=}")
# TODO is there a clean way to hang on to a session for a bit when the cli command has run?
# i guess realistically we don't want that
args.func(config, args)

View File

@@ -0,0 +1,168 @@
import argparse
import logging
import typing
import pathlib
import taiga_pycli.config
import taiga_pycli.service
import taiga_pycli.workout_utils
from taiga_pycli.exceptions import (
AuthenticationError,
ValidationError,
NetworkError,
TryGoAPIError,
)
_logger = logging.getLogger(__name__)
if typing.TYPE_CHECKING:
_SubparserType = argparse._SubParsersAction[argparse.ArgumentParser]
else:
_SubparserType = typing.Any
def setup_parser(subparsers: _SubparserType) -> None:
"""Setup the activities command group with its subcommands"""
parser = subparsers.add_parser("activities", help="Manage activity files")
activities_subparsers = parser.add_subparsers(dest="activities_cmd", required=True)
# Upload subcommand
upload_parser = activities_subparsers.add_parser(
"upload", help="Upload an activity file to the server"
)
upload_parser.add_argument(
"file_path", type=pathlib.Path, help="Path to the activity file to upload"
)
upload_parser.set_defaults(func=run_upload)
# List subcommand
list_parser = activities_subparsers.add_parser(
"ls", help="List your activity files"
)
list_parser.set_defaults(func=run_list)
# Download subcommand
download_parser = activities_subparsers.add_parser(
"download", help="Download an activity file by ID"
)
download_parser.add_argument("id", type=int, help="Activity file ID to download")
download_parser.add_argument(
"--output",
"-o",
type=pathlib.Path,
help="Output file path (defaults to original filename)",
)
download_parser.set_defaults(func=run_download)
def run_upload(config: taiga_pycli.config.Config, args):
"""Run the upload command"""
try:
clnt = taiga_pycli.service.BackendService(config)
_logger.info(f"Uploading file: {args.file_path}")
activity_file = clnt.upload_activity_file(args.file_path)
print("Upload successful!")
print(f"File ID: {activity_file.id}")
print(f"Timestamp: {activity_file.timestamp}")
print(f"Created at: {activity_file.created_at}")
except AuthenticationError as e:
print(f"Authentication error: {e}")
print("Please run 'taiga login' first.")
return 1
except ValidationError as e:
print(f"Validation error: {e}")
return 1
except NetworkError as e:
print(f"Network error: {e}")
return 1
except TryGoAPIError as e:
print(f"API error: {e}")
return 1
except Exception as e:
_logger.error(f"Unexpected error during upload: {e}")
print(f"Unexpected error: {e}")
return 1
return 0
def run_list(config: taiga_pycli.config.Config, args):
"""Run the list command"""
try:
clnt = taiga_pycli.service.BackendService(config)
_logger.info("Listing activity files")
activity_files = clnt.list_activity_files()
if not activity_files:
print("No activity files found.")
return 0
# Print header
print(f"{'ID':<5} {'Timestamp':<20} {'Created':<20}")
print("-" * 45)
# Print each activity file
for activity_file in activity_files:
timestamp_formatted = taiga_pycli.workout_utils.format_activity_timestamp(activity_file.timestamp)
created_formatted = taiga_pycli.workout_utils.format_activity_timestamp(activity_file.created_at)
print(
f"{activity_file.id:<5} {timestamp_formatted:<20} {created_formatted:<20}"
)
except AuthenticationError as e:
print(f"Authentication error: {e}")
print("Please run 'taiga login' first.")
return 1
except NetworkError as e:
print(f"Network error: {e}")
return 1
except TryGoAPIError as e:
print(f"API error: {e}")
return 1
except Exception as e:
_logger.error(f"Unexpected error during list: {e}")
print(f"Unexpected error: {e}")
return 1
return 0
def run_download(config: taiga_pycli.config.Config, args):
"""Run the download command"""
try:
clnt = taiga_pycli.service.BackendService(config)
_logger.info(f"Downloading activity file ID: {args.id}")
output_path = clnt.download_activity_file(args.id, args.output)
print("Download successful!")
print(f"File saved to: {output_path}")
except AuthenticationError as e:
print(f"Authentication error: {e}")
print("Please run 'taiga login' first.")
return 1
except ValidationError as e:
print(f"Validation error: {e}")
return 1
except NetworkError as e:
print(f"Network error: {e}")
return 1
except TryGoAPIError as e:
print(f"API error: {e}")
return 1
except Exception as e:
_logger.error(f"Unexpected error during download: {e}")
print(f"Unexpected error: {e}")
return 1
return 0

View File

@@ -1,11 +1,11 @@
import trygo_py_cliclient.config import taiga_pycli.config
import typing import typing
import logging import logging
import pathlib import pathlib
def set_up_logging( def set_up_logging(
config: trygo_py_cliclient.config.Config, config: taiga_pycli.config.Config,
create_logfile_parents: bool = True, create_logfile_parents: bool = True,
): ):
# for convenience # for convenience

View File

@@ -0,0 +1,66 @@
import argparse
import logging
import typing
import taiga_pycli.config
import taiga_pycli.models
import taiga_pycli.service
_logger = logging.getLogger(__name__)
if typing.TYPE_CHECKING:
_SubparserType = argparse._SubParsersAction[argparse.ArgumentParser]
else:
_SubparserType = typing.Any
def setup_parser(subparsers: _SubparserType) -> None:
parser = subparsers.add_parser("hat")
parser.add_argument(
"--name", type=str, help="The name of the hat to add", default=None
)
parser.add_argument(
"--description",
type=str,
help="The description of the hat to add",
default=None,
)
parser.set_defaults(func=run)
def run(cfg: taiga_pycli.config.Config, args):
# clnt = taiga_pycli.client.TryGoAPIClient("http://localhost:8080/")
backend = taiga_pycli.service.BackendService(cfg)
if args.name is not None:
if args.description is None:
_logger.error("Got a null description, exiting")
return
# both not None
backend.add_hat(args.name, args.description)
return
else:
_logger.debug("Not creating, just list")
if args.description is not None:
_logger.error("Provided a description without name")
response = backend.get_hats()
if response is None:
_logger.warning("none response here")
return
real_hats = []
for hat in response:
rh = taiga_pycli.models.Hat(
name=hat["name"],
description=hat["description"],
user_id=hat["user_id"],
)
real_hats.append(rh)
print(rh)
# _logger.info(response)
return

View File

@@ -0,0 +1,52 @@
import argparse
import logging
import typing
import getpass
import taiga_pycli.config
import taiga_pycli.service
_logger = logging.getLogger(__name__)
if typing.TYPE_CHECKING:
_SubparserType = argparse._SubParsersAction[argparse.ArgumentParser]
else:
_SubparserType = typing.Any
class Password:
DEFAULT = "Prompt if not provided"
def __init__(self, value):
if value == self.DEFAULT:
value = getpass.getpass("Password: ")
self.value = value
def __str__(self):
return self.value
def setup_parser(subparsers: _SubparserType) -> None:
parser = subparsers.add_parser("login")
parser.add_argument(
"--email", type=str, help="The email to log in with ", default=None
)
parser.add_argument(
"--password", type=Password, help="Password", default=Password.DEFAULT
)
parser.set_defaults(func=run)
def run(config: taiga_pycli.config.Config, args):
clnt = taiga_pycli.service.BackendService(config)
_logger.info(f"using password {args.password}")
email_to_use = args.email
if args.email is None:
email_to_use = config.general_config.backend_user
clnt.login(email_to_use, str(args.password))
# _logger.info(response)
return

View File

@@ -1,8 +1,9 @@
import argparse import argparse
import logging import logging
import typing import typing
import taiga_pycli.config
import trygo_py_cliclient.client import taiga_pycli.service
_logger = logging.getLogger(__name__) _logger = logging.getLogger(__name__)
@@ -24,7 +25,9 @@ def setup_parser(subparsers: _SubparserType) -> None:
parser.set_defaults(func=run) parser.set_defaults(func=run)
def run(args): def run(cfg: taiga_pycli.config.Config, args):
clnt = trygo_py_cliclient.client.TryGoAPIClient("http://localhost:8080/") # clnt = taiga_pycli.client.TryGoAPIClient("http://localhost:8080/")
response = clnt.register(args.email, args.password, args.display_name)
backend = taiga_pycli.service.BackendService(cfg)
response = backend.register(args.email, args.password, args.display_name)
_logger.info(response) _logger.info(response)

View File

@@ -0,0 +1,68 @@
import argparse
import logging
import typing
import pathlib
import taiga_pycli.config
import taiga_pycli.service
from taiga_pycli.exceptions import (
AuthenticationError,
ValidationError,
NetworkError,
TryGoAPIError,
)
_logger = logging.getLogger(__name__)
if typing.TYPE_CHECKING:
_SubparserType = argparse._SubParsersAction[argparse.ArgumentParser]
else:
_SubparserType = typing.Any
def setup_parser(subparsers: _SubparserType) -> None:
parser = subparsers.add_parser(
"upload", help="Upload an activity file to the server"
)
parser.add_argument(
"file_path", type=pathlib.Path, help="Path to the activity file to upload"
)
parser.set_defaults(func=run)
def run(config: taiga_pycli.config.Config, args):
"""Run the upload command"""
try:
clnt = taiga_pycli.service.BackendService(config)
_logger.info(f"Uploading file: {args.file_path}")
activity_file = clnt.upload_activity_file(args.file_path)
print("Upload successful!")
print(f"File ID: {activity_file.id}")
print(f"Timestamp: {activity_file.timestamp}")
print(f"Created at: {activity_file.created_at}")
except AuthenticationError as e:
print(f"Authentication error: {e}")
print("Please run 'taiga login' first.")
return 1
except ValidationError as e:
print(f"Validation error: {e}")
return 1
except NetworkError as e:
print(f"Network error: {e}")
return 1
except TryGoAPIError as e:
print(f"API error: {e}")
return 1
except Exception as e:
_logger.error(f"Unexpected error during upload: {e}")
print(f"Unexpected error: {e}")
return 1
return 0

View File

@@ -0,0 +1,299 @@
import argparse
import logging
import typing
import pathlib
import taiga_pycli.config
import taiga_pycli.service
import taiga_pycli.workout_utils
from taiga_pycli.models import CreateWorkoutRequest
from taiga_pycli.exceptions import (
AuthenticationError,
ValidationError,
NetworkError,
TryGoAPIError,
)
_logger = logging.getLogger(__name__)
if typing.TYPE_CHECKING:
_SubparserType = argparse._SubParsersAction[argparse.ArgumentParser]
else:
_SubparserType = typing.Any
def setup_parser(subparsers: _SubparserType) -> None:
"""Setup the workouts command group with its subcommands"""
parser = subparsers.add_parser("workouts", help="Manage workouts")
workouts_subparsers = parser.add_subparsers(dest="workouts_cmd", required=True)
# Upload subcommand - primary workflow
upload_parser = workouts_subparsers.add_parser(
"upload", help="Upload a FIT file and create workout automatically"
)
upload_parser.add_argument(
"file_path", type=pathlib.Path, help="Path to the FIT file to upload"
)
upload_parser.set_defaults(func=run_upload)
# List subcommand
list_parser = workouts_subparsers.add_parser(
"ls", help="List your workouts"
)
list_parser.set_defaults(func=run_list)
# Show subcommand
show_parser = workouts_subparsers.add_parser(
"show", help="Show detailed workout information"
)
show_parser.add_argument("id", type=int, help="Workout ID to show")
show_parser.set_defaults(func=run_show)
# Create subcommand for manual workout entry
create_parser = workouts_subparsers.add_parser(
"create", help="Create a workout manually"
)
create_parser.add_argument("--distance", type=float, help="Distance in miles")
create_parser.add_argument("--time", type=int, help="Duration in seconds")
create_parser.add_argument("--pace", type=float, help="Pace in minutes per mile")
create_parser.add_argument("--speed", type=float, help="Speed in mph")
create_parser.add_argument("--start-time", type=str, help="Start time (ISO format)")
create_parser.add_argument("--end-time", type=str, help="End time (ISO format)")
create_parser.set_defaults(func=run_create)
# Delete subcommand
delete_parser = workouts_subparsers.add_parser(
"delete", help="Delete a workout"
)
delete_parser.add_argument("id", type=int, help="Workout ID to delete")
delete_parser.add_argument("--yes", "-y", action="store_true", help="Skip confirmation")
delete_parser.set_defaults(func=run_delete)
def run_upload(config: taiga_pycli.config.Config, args):
"""Run the upload command - primary workflow"""
try:
clnt = taiga_pycli.service.BackendService(config)
_logger.info(f"Uploading FIT file and creating workout: {args.file_path}")
# Check file extension
if not args.file_path.name.lower().endswith('.fit'):
print(f"Warning: File does not have .fit extension: {args.file_path}")
# Upload and create workout in one step
workout = clnt.upload_and_create_workout(args.file_path)
print("Workout created successfully!")
print(f"Workout ID: {workout.id}")
# Print formatted workout summary
summary = taiga_pycli.workout_utils.format_workout_summary(workout)
if summary != "Workout data":
print(f"Summary: {summary}")
# Show additional details if available
if workout.distance_miles or workout.time_seconds:
print("\nWorkout Details:")
taiga_pycli.workout_utils.print_workout_details(workout)
except AuthenticationError as e:
print(f"Authentication error: {e}")
print("Please run 'taiga login' first.")
return 1
except ValidationError as e:
print(f"Validation error: {e}")
return 1
except NetworkError as e:
print(f"Network error: {e}")
return 1
except TryGoAPIError as e:
print(f"API error: {e}")
return 1
except Exception as e:
_logger.error(f"Unexpected error during workout upload: {e}")
print(f"Unexpected error: {e}")
return 1
return 0
def run_list(config: taiga_pycli.config.Config, args):
"""Run the list command"""
try:
clnt = taiga_pycli.service.BackendService(config)
_logger.info("Listing workouts")
workouts = clnt.get_workouts()
if not workouts:
print("No workouts found.")
print("Upload a FIT file with 'taiga workouts upload <file.fit>' to get started!")
return 0
# Print table header
taiga_pycli.workout_utils.print_workout_table_header()
# Print each workout
for workout in workouts:
taiga_pycli.workout_utils.print_workout_row(workout)
print(f"\nTotal: {len(workouts)} workouts")
except AuthenticationError as e:
print(f"Authentication error: {e}")
print("Please run 'taiga login' first.")
return 1
except NetworkError as e:
print(f"Network error: {e}")
return 1
except TryGoAPIError as e:
print(f"API error: {e}")
return 1
except Exception as e:
_logger.error(f"Unexpected error during list: {e}")
print(f"Unexpected error: {e}")
return 1
return 0
def run_show(config: taiga_pycli.config.Config, args):
"""Run the show command"""
try:
clnt = taiga_pycli.service.BackendService(config)
_logger.info(f"Showing workout {args.id}")
workout = clnt.get_workout(args.id)
# Print detailed workout information
taiga_pycli.workout_utils.print_workout_details(workout)
except AuthenticationError as e:
print(f"Authentication error: {e}")
print("Please run 'taiga login' first.")
return 1
except ValidationError as e:
print(f"Workout not found: {e}")
return 1
except NetworkError as e:
print(f"Network error: {e}")
return 1
except TryGoAPIError as e:
print(f"API error: {e}")
return 1
except Exception as e:
_logger.error(f"Unexpected error during show: {e}")
print(f"Unexpected error: {e}")
return 1
return 0
def run_create(config: taiga_pycli.config.Config, args):
"""Run the create command for manual workout entry"""
try:
clnt = taiga_pycli.service.BackendService(config)
_logger.info("Creating manual workout")
# Build request from args
request = CreateWorkoutRequest(
distance_miles=args.distance,
time_seconds=args.time,
pace_min_per_mile=args.pace,
speed_mph=args.speed,
start_time=args.start_time,
end_time=args.end_time,
)
# Validate that at least some data is provided
if not any([args.distance, args.time, args.pace, args.speed, args.start_time]):
print("Error: At least one workout parameter must be provided")
print("Use --distance, --time, --pace, --speed, or --start-time")
return 1
workout = clnt.create_workout(request)
print("Workout created successfully!")
print(f"Workout ID: {workout.id}")
# Show workout details
print("\nWorkout Details:")
taiga_pycli.workout_utils.print_workout_details(workout)
except AuthenticationError as e:
print(f"Authentication error: {e}")
print("Please run 'taiga login' first.")
return 1
except ValidationError as e:
print(f"Validation error: {e}")
return 1
except NetworkError as e:
print(f"Network error: {e}")
return 1
except TryGoAPIError as e:
print(f"API error: {e}")
return 1
except Exception as e:
_logger.error(f"Unexpected error during create: {e}")
print(f"Unexpected error: {e}")
return 1
return 0
def run_delete(config: taiga_pycli.config.Config, args):
"""Run the delete command"""
try:
clnt = taiga_pycli.service.BackendService(config)
# Get workout details first to show what we're deleting
try:
workout = clnt.get_workout(args.id)
except (ValidationError, TryGoAPIError):
print(f"Workout {args.id} not found.")
return 1
# Show what we're about to delete
print(f"About to delete workout #{workout.id}:")
summary = taiga_pycli.workout_utils.format_workout_summary(workout)
date = taiga_pycli.workout_utils.format_workout_date(workout.start_time or workout.created_at)
print(f" {date}: {summary}")
# Confirm deletion unless --yes flag is used
if not args.yes:
response = input("\nAre you sure you want to delete this workout? (y/N): ")
if response.lower() not in ['y', 'yes']:
print("Deletion cancelled.")
return 0
_logger.info(f"Deleting workout {args.id}")
clnt.delete_workout(args.id)
print(f"Workout {args.id} deleted successfully.")
except AuthenticationError as e:
print(f"Authentication error: {e}")
print("Please run 'taiga login' first.")
return 1
except ValidationError as e:
print(f"Workout not found: {e}")
return 1
except NetworkError as e:
print(f"Network error: {e}")
return 1
except TryGoAPIError as e:
print(f"API error: {e}")
return 1
except Exception as e:
_logger.error(f"Unexpected error during delete: {e}")
print(f"Unexpected error: {e}")
return 1
return 0

View File

@@ -0,0 +1,10 @@
from taiga_pycli.config.config import (
GeneralConfig,
Config,
)
from taiga_pycli.config.config_reader import (
read_config,
)
__all__ = ["GeneralConfig", "Config", "read_config"]

View File

@@ -3,7 +3,7 @@ import dacite
import pathlib import pathlib
import tomli import tomli
from trygo_py_cliclient.config import GeneralConfig, Config from taiga_pycli.config import GeneralConfig, Config
_logger = logging.getLogger(__name__) _logger = logging.getLogger(__name__)

102
src/taiga_pycli/models.py Normal file
View File

@@ -0,0 +1,102 @@
from dataclasses import dataclass
from typing import Optional
@dataclass
class User:
"""User model for API requests and responses"""
email: str
password: str
display_name: str
id: Optional[str] = None
@dataclass
class RegisterRequest:
"""Request payload for /auth/register endpoint"""
email: str
password: str
display_name: str
@dataclass
class LoginRequest:
"""Request payload for /auth/login endpoint"""
email: str
password: str
@dataclass
class AuthResponse:
"""Response from authentication endpoints"""
message: str
email: Optional[str] = None
id: Optional[str] = None
@dataclass
class Hat:
"""A wearable hat"""
name: str
description: str
user_id: int
@dataclass
class ActivityFile:
"""Activity file model for API responses"""
id: int
timestamp: str
file_repo_hash: Optional[str]
created_at: str
updated_at: str
user_id: int
@dataclass
class Workout:
"""Workout model for API responses"""
id: int
distance_miles: Optional[float]
time_seconds: Optional[float]
speed_mph: Optional[float]
pace_min_per_mile: Optional[float]
start_time: Optional[str]
end_time: Optional[str]
activity_file_id: Optional[int]
created_at: str
updated_at: str
user_id: int
@dataclass
class CreateWorkoutRequest:
"""Request payload for creating a new workout"""
distance_miles: Optional[float] = None
time_seconds: Optional[float] = None
speed_mph: Optional[float] = None
pace_min_per_mile: Optional[float] = None
start_time: Optional[str] = None
end_time: Optional[str] = None
activity_file_id: Optional[int] = None
@dataclass
class UpdateWorkoutRequest:
"""Request payload for updating an existing workout"""
distance_miles: Optional[float] = None
time_seconds: Optional[float] = None
speed_mph: Optional[float] = None
pace_min_per_mile: Optional[float] = None
start_time: Optional[str] = None
end_time: Optional[str] = None
activity_file_id: Optional[int] = None

View File

@@ -0,0 +1,718 @@
import logging
from typing import Optional, Dict, Any, Sequence, List
from dataclasses import asdict
from pathlib import Path
import requests
from taiga_pycli.models import (
RegisterRequest,
LoginRequest,
AuthResponse,
ActivityFile,
Workout,
CreateWorkoutRequest,
UpdateWorkoutRequest,
)
from taiga_pycli.config import Config
from taiga_pycli.exceptions import (
TryGoAPIError,
AuthenticationError,
ValidationError,
NetworkError,
ServerError,
)
_logger = logging.getLogger(__name__)
class BackendService:
def __init__(self, config: Config):
self.base_url = config.general_config.backend_base_url.rstrip("/")
self.timeout = 30
self.session = requests.Session()
self.session.headers.update(
{"Content-Type": "application/json", "User-Agent": "trygo-py-client/0.1.0"}
)
self._token_path = Path("cred") / "token"
self.token: Optional[str] = None
def _make_request(
self, method: str, endpoint: str, data: Optional[Dict[str, Any]] = None
) -> requests.Response:
"""
Make HTTP request to API endpoint
Args:
method: HTTP method (GET, POST, PUT, DELETE)
endpoint: API endpoint path
data: Request payload data
Returns:
Response data as dictionary
Raises:
NetworkError: For connection/timeout errors
AuthenticationError: For 401 errors
ValidationError: For 400 errors
ServerError: For 5xx errors
TryGoAPIError: For other API errors
"""
url = f"{self.base_url}{endpoint}"
_logger.info(f"Making {method} request to {url}")
try:
response = self.session.request(
method=method, url=url, json=data, timeout=self.timeout
)
# Log response status
_logger.debug(f"Response status: {response.status_code}")
# Handle different HTTP status codes
if response.status_code == 401:
raise AuthenticationError("Authentication failed")
elif response.status_code == 400:
error_msg = "Validation error"
try:
error_data = response.json()
if "message" in error_data:
error_msg = error_data["message"]
except Exception:
pass
raise ValidationError(error_msg)
elif 500 <= response.status_code < 600:
raise ServerError(f"Server error: {response.status_code}")
elif not response.ok:
raise TryGoAPIError(f"API error: {response.status_code}")
# Parse JSON response (skip for 204 No Content)
if response.status_code == 204:
return response
try:
_logger.debug(response.json())
return response
except ValueError as e:
raise TryGoAPIError(f"Invalid JSON response: {e}")
# claude really fucked this
except requests.exceptions.Timeout:
raise NetworkError("Request timeout")
except requests.exceptions.ConnectionError:
raise NetworkError("Connection error")
except requests.exceptions.RequestException as e:
raise NetworkError(f"Network error: {e}")
def register(self, email: str, password: str, display_name: str) -> AuthResponse:
"""
Register a new user account
"""
request_data = RegisterRequest(
email=email, password=password, display_name=display_name
)
_logger.info(f"Registering user: {email}")
response = self._make_request("POST", "/auth/register", asdict(request_data))
response_json = response.json()
# Parse response and create AuthResponse
auth_response = AuthResponse(
message=response_json.get("message", ""),
email=response_json.get("email", ""),
id=response_json.get("id", ""),
)
return auth_response
def login(self, email: str, password: str) -> str:
"""
Authenticate user and get auth token
"""
request_data = LoginRequest(email=email, password=password)
_logger.info(f"Logging in user: {email}")
response = self._make_request("POST", "/auth/tokens", asdict(request_data))
response_data = response.json()
_logger.info(response_data)
# Parse response and create AuthResponse
token = response_data.get("token", "")
self._store_credential(token)
return token
def logout(self) -> None:
"""
Clear authentication token and session data
Optionally calls logout endpoint if server supports it
"""
_logger.info("Logging out user")
# Try to call logout endpoint if we have a token
_logger.debug("Local authentication cleared")
def get_hats(self) -> Optional[Sequence[Dict[str, Any]]]:
# needs credential
cred = self._read_credential()
if cred is None:
return None
_logger.debug("Credential was read")
response = self._make_request("GET", "/hats")
response_data = response.json()
# _logger.debug(response)
return response_data
def add_hat(self, name: str, description: str) -> Optional[Dict[str, Any]]:
cred = self._read_credential()
if cred is None:
return None
_logger.debug("Credential was read")
response = self._make_request(
"POST", "/hats", {"name": name, "description": description}
)
return response.json()
def _store_credential(self, token: str) -> None:
self._token_path.parent.mkdir(parents=True, exist_ok=True)
with self._token_path.open(mode="w") as tokenf:
tokenf.write(token)
self.token = token
self.session.headers.update({"Authorization": f"Bearer {token}"})
def _read_credential(self) -> Optional[str]:
try:
with open(self._token_path) as token_file:
token_in_file = token_file.read()
self.token = token_in_file
self.session.headers.update(
{"Authorization": f"Bearer {token_in_file}"}
)
return token_in_file
except FileNotFoundError:
_logger.error("No token file found, try logging in")
return None
def upload_activity_file(self, file_path: Path) -> ActivityFile:
"""
Upload an activity file to the server
Args:
file_path: Path to the file to upload
Returns:
ActivityFile object with server response data
Raises:
AuthenticationError: If not authenticated
ValidationError: If file is invalid
NetworkError: For connection issues
"""
cred = self._read_credential()
if cred is None:
raise AuthenticationError("Not authenticated. Please login first.")
if not file_path.exists():
raise ValidationError(f"File not found: {file_path}")
if not file_path.is_file():
raise ValidationError(f"Path is not a file: {file_path}")
_logger.info(f"Uploading activity file: {file_path}")
url = f"{self.base_url}/activity_files"
try:
with open(file_path, "rb") as file_obj:
files = {"file": (file_path.name, file_obj, "application/octet-stream")}
# Temporarily remove Content-Type header to let requests set multipart boundary
original_content_type = self.session.headers.pop("Content-Type", None)
try:
response = self.session.post(
url=url, files=files, timeout=self.timeout
)
finally:
# Restore original Content-Type header
if original_content_type:
self.session.headers["Content-Type"] = original_content_type
# Handle response status codes
if response.status_code == 401:
raise AuthenticationError("Authentication failed")
elif response.status_code == 400:
error_msg = "File upload failed"
try:
error_data = response.json()
if "error" in error_data:
error_msg = error_data["error"]
except Exception:
pass
raise ValidationError(error_msg)
elif 500 <= response.status_code < 600:
raise ServerError(f"Server error: {response.status_code}")
elif not response.ok:
raise TryGoAPIError(f"Upload failed: {response.status_code}")
# Parse response
try:
response_data = response.json()
_logger.info(f"Upload successful: {response_data}")
return ActivityFile(
id=response_data["id"],
timestamp=response_data["timestamp"],
file_repo_hash=response_data.get("file_repo_hash"),
created_at=response_data["created_at"],
updated_at=response_data["updated_at"],
user_id=response_data["user_id"],
)
except ValueError as e:
raise TryGoAPIError(f"Invalid JSON response: {e}")
except requests.exceptions.Timeout:
raise NetworkError("Upload timeout")
except requests.exceptions.ConnectionError:
raise NetworkError("Connection error during upload")
except requests.exceptions.RequestException as e:
raise NetworkError(f"Network error during upload: {e}")
except FileNotFoundError:
raise ValidationError(f"File not found: {file_path}")
except PermissionError:
raise ValidationError(f"Permission denied reading file: {file_path}")
def list_activity_files(self) -> List[ActivityFile]:
"""
List all activity files for the authenticated user
Returns:
List of ActivityFile objects
Raises:
AuthenticationError: If not authenticated
NetworkError: For connection issues
TryGoAPIError: For API errors
"""
cred = self._read_credential()
if cred is None:
raise AuthenticationError("Not authenticated. Please login first.")
_logger.info("Fetching activity files list")
try:
response = self._make_request("GET", "/activity_files")
response_data = response.json()
# Parse response into ActivityFile objects
_logger.debug(response_data)
activity_files: List[ActivityFile] = []
_logger.debug("checking if list")
if isinstance(response_data, list):
_logger.debug("yes in list")
for item in response_data:
_logger.debug(item)
if isinstance(item, dict):
activity_file = ActivityFile(
id=item["id"],
timestamp=item["timestamp"],
file_repo_hash=item.get("file_repo_hash"),
created_at=item["created_at"],
updated_at=item["updated_at"],
user_id=item["user_id"],
)
activity_files.append(activity_file)
_logger.info(f"Retrieved {len(activity_files)} activity files")
return activity_files
except requests.exceptions.Timeout:
raise NetworkError("Request timeout")
except requests.exceptions.ConnectionError:
raise NetworkError("Connection error")
except requests.exceptions.RequestException as e:
raise NetworkError(f"Network error: {e}")
def download_activity_file(
self, file_id: int, output_path: Optional[Path] = None
) -> Path:
"""
Download an activity file by ID
Args:
file_id: The ID of the activity file to download
output_path: Optional path to save the file (defaults to activity_{id}.fit)
Returns:
Path where the file was saved
Raises:
AuthenticationError: If not authenticated
ValidationError: If file not found or invalid
NetworkError: For connection issues
TryGoAPIError: For API errors
"""
cred = self._read_credential()
if cred is None:
raise AuthenticationError("Not authenticated. Please login first.")
if output_path is None:
output_path = Path(f"activity_{file_id}.fit")
_logger.info(f"Downloading activity file {file_id} to {output_path}")
url = f"{self.base_url}/activity_files/{file_id}/download"
try:
response = self.session.get(url=url, timeout=self.timeout, stream=True)
# Handle response status codes
if response.status_code == 401:
raise AuthenticationError("Authentication failed")
elif response.status_code == 404:
raise ValidationError(f"Activity file with ID {file_id} not found")
elif response.status_code == 400:
raise ValidationError("Invalid download request")
elif 500 <= response.status_code < 600:
raise ServerError(f"Server error: {response.status_code}")
elif not response.ok:
raise TryGoAPIError(f"Download failed: {response.status_code}")
# Write file to disk
with open(output_path, "wb") as f:
for chunk in response.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
_logger.info(f"Downloaded activity file {file_id} to {output_path}")
return output_path
except requests.exceptions.Timeout:
raise NetworkError("Download timeout")
except requests.exceptions.ConnectionError:
raise NetworkError("Connection error during download")
except requests.exceptions.RequestException as e:
raise NetworkError(f"Network error during download: {e}")
except OSError as e:
raise ValidationError(f"Error writing file {output_path}: {e}")
def upload_and_create_workout(self, file_path: Path) -> Workout:
"""
Upload FIT file and automatically create workout in one operation
Args:
file_path: Path to the FIT file to upload
Returns:
Workout object created from the FIT file
Raises:
AuthenticationError: If not authenticated
ValidationError: If file is invalid or FIT processing fails
NetworkError: For connection issues
TryGoAPIError: For API errors
"""
_logger.info(f"Uploading FIT file and creating workout: {file_path}")
activity_file = self.upload_activity_file(file_path)
_logger.info(f"Activity file uploaded with ID: {activity_file.id}")
try:
workout = self.create_workout_from_activity_file(activity_file.id)
_logger.info(f"Workout created with ID: {workout.id}")
return workout
except Exception as e:
_logger.error(f"Failed to create workout from activity file {activity_file.id}: {e}")
raise
def get_workouts(self) -> List[Workout]:
"""
Get all workouts for the authenticated user
Returns:
List of Workout objects
Raises:
AuthenticationError: If not authenticated
NetworkError: For connection issues
TryGoAPIError: For API errors
"""
cred = self._read_credential()
if cred is None:
raise AuthenticationError("Not authenticated. Please login first.")
_logger.info("Fetching workouts list")
try:
response = self._make_request("GET", "/workouts")
response_data = response.json()
workouts: List[Workout] = []
if isinstance(response_data, list):
for item in response_data:
if isinstance(item, dict):
workout = Workout(
id=item["id"],
distance_miles=item.get("distance_miles"),
time_seconds=item.get("time_seconds"),
speed_mph=item.get("speed_mph"),
pace_min_per_mile=item.get("pace_min_per_mile"),
start_time=item.get("start_time"),
end_time=item.get("end_time"),
activity_file_id=item.get("activity_file_id"),
created_at=item["created_at"],
updated_at=item["updated_at"],
user_id=item["user_id"],
)
workouts.append(workout)
_logger.info(f"Retrieved {len(workouts)} workouts")
return workouts
except requests.exceptions.Timeout:
raise NetworkError("Request timeout")
except requests.exceptions.ConnectionError:
raise NetworkError("Connection error")
except requests.exceptions.RequestException as e:
raise NetworkError(f"Network error: {e}")
def get_workout(self, workout_id: int) -> Workout:
"""
Get a specific workout by ID
Args:
workout_id: The ID of the workout to retrieve
Returns:
Workout object
Raises:
AuthenticationError: If not authenticated
ValidationError: If workout not found
NetworkError: For connection issues
TryGoAPIError: For API errors
"""
cred = self._read_credential()
if cred is None:
raise AuthenticationError("Not authenticated. Please login first.")
_logger.info(f"Fetching workout {workout_id}")
try:
response = self._make_request("GET", f"/workouts/{workout_id}")
item = response.json()
workout = Workout(
id=item["id"],
distance_miles=item.get("distance_miles"),
time_seconds=item.get("time_seconds"),
speed_mph=item.get("speed_mph"),
pace_min_per_mile=item.get("pace_min_per_mile"),
start_time=item.get("start_time"),
end_time=item.get("end_time"),
activity_file_id=item.get("activity_file_id"),
created_at=item["created_at"],
updated_at=item["updated_at"],
user_id=item["user_id"],
)
return workout
except requests.exceptions.Timeout:
raise NetworkError("Request timeout")
except requests.exceptions.ConnectionError:
raise NetworkError("Connection error")
except requests.exceptions.RequestException as e:
raise NetworkError(f"Network error: {e}")
def create_workout(self, request: CreateWorkoutRequest) -> Workout:
"""
Create a new workout manually
Args:
request: CreateWorkoutRequest with workout data
Returns:
Created Workout object
Raises:
AuthenticationError: If not authenticated
ValidationError: If request data is invalid
NetworkError: For connection issues
TryGoAPIError: For API errors
"""
cred = self._read_credential()
if cred is None:
raise AuthenticationError("Not authenticated. Please login first.")
_logger.info("Creating new workout")
try:
response = self._make_request("POST", "/workouts", asdict(request))
item = response.json()
workout = Workout(
id=item["id"],
distance_miles=item.get("distance_miles"),
time_seconds=item.get("time_seconds"),
speed_mph=item.get("speed_mph"),
pace_min_per_mile=item.get("pace_min_per_mile"),
start_time=item.get("start_time"),
end_time=item.get("end_time"),
activity_file_id=item.get("activity_file_id"),
created_at=item["created_at"],
updated_at=item["updated_at"],
user_id=item["user_id"],
)
_logger.info(f"Created workout with ID: {workout.id}")
return workout
except requests.exceptions.Timeout:
raise NetworkError("Request timeout")
except requests.exceptions.ConnectionError:
raise NetworkError("Connection error")
except requests.exceptions.RequestException as e:
raise NetworkError(f"Network error: {e}")
def update_workout(self, workout_id: int, request: UpdateWorkoutRequest) -> Workout:
"""
Update an existing workout
Args:
workout_id: The ID of the workout to update
request: UpdateWorkoutRequest with fields to update
Returns:
Updated Workout object
Raises:
AuthenticationError: If not authenticated
ValidationError: If workout not found or request data is invalid
NetworkError: For connection issues
TryGoAPIError: For API errors
"""
cred = self._read_credential()
if cred is None:
raise AuthenticationError("Not authenticated. Please login first.")
_logger.info(f"Updating workout {workout_id}")
try:
response = self._make_request("PUT", f"/workouts/{workout_id}", asdict(request))
item = response.json()
workout = Workout(
id=item["id"],
distance_miles=item.get("distance_miles"),
time_seconds=item.get("time_seconds"),
speed_mph=item.get("speed_mph"),
pace_min_per_mile=item.get("pace_min_per_mile"),
start_time=item.get("start_time"),
end_time=item.get("end_time"),
activity_file_id=item.get("activity_file_id"),
created_at=item["created_at"],
updated_at=item["updated_at"],
user_id=item["user_id"],
)
_logger.info(f"Updated workout {workout_id}")
return workout
except requests.exceptions.Timeout:
raise NetworkError("Request timeout")
except requests.exceptions.ConnectionError:
raise NetworkError("Connection error")
except requests.exceptions.RequestException as e:
raise NetworkError(f"Network error: {e}")
def delete_workout(self, workout_id: int) -> None:
"""
Delete a workout
Args:
workout_id: The ID of the workout to delete
Raises:
AuthenticationError: If not authenticated
ValidationError: If workout not found
NetworkError: For connection issues
TryGoAPIError: For API errors
"""
cred = self._read_credential()
if cred is None:
raise AuthenticationError("Not authenticated. Please login first.")
_logger.info(f"Deleting workout {workout_id}")
try:
self._make_request("DELETE", f"/workouts/{workout_id}")
# DELETE returns 204 No Content with empty body, so don't try to parse JSON
_logger.info(f"Deleted workout {workout_id}")
except requests.exceptions.Timeout:
raise NetworkError("Request timeout")
except requests.exceptions.ConnectionError:
raise NetworkError("Connection error")
except requests.exceptions.RequestException as e:
raise NetworkError(f"Network error: {e}")
def create_workout_from_activity_file(self, activity_file_id: int) -> Workout:
"""
Create workout from an existing activity file
Args:
activity_file_id: The ID of the activity file to process
Returns:
Created Workout object
Raises:
AuthenticationError: If not authenticated
ValidationError: If activity file not found or FIT processing fails
NetworkError: For connection issues
TryGoAPIError: For API errors
"""
cred = self._read_credential()
if cred is None:
raise AuthenticationError("Not authenticated. Please login first.")
_logger.info(f"Creating workout from activity file {activity_file_id}")
try:
response = self._make_request(
"POST", f"/workouts/from-activity-file/{activity_file_id}"
)
item = response.json()
workout = Workout(
id=item["id"],
distance_miles=item.get("distance_miles"),
time_seconds=item.get("time_seconds"),
speed_mph=item.get("speed_mph"),
pace_min_per_mile=item.get("pace_min_per_mile"),
start_time=item.get("start_time"),
end_time=item.get("end_time"),
activity_file_id=item.get("activity_file_id"),
created_at=item["created_at"],
updated_at=item["updated_at"],
user_id=item["user_id"],
)
_logger.info(f"Created workout {workout.id} from activity file {activity_file_id}")
return workout
except requests.exceptions.Timeout:
raise NetworkError("Request timeout")
except requests.exceptions.ConnectionError:
raise NetworkError("Connection error")
except requests.exceptions.RequestException as e:
raise NetworkError(f"Network error: {e}")

View File

@@ -0,0 +1,199 @@
"""Utility functions for workout data formatting and display"""
from typing import Optional, Union
from datetime import datetime
from taiga_pycli.models import Workout
def format_time_duration(seconds: Optional[Union[int, float]]) -> str:
"""Format time duration in seconds to HH:MM:SS or MM:SS"""
if seconds is None:
return "--:--"
# Convert to int to handle float values from API
total_seconds = int(seconds)
hours = total_seconds // 3600
minutes = (total_seconds % 3600) // 60
secs = total_seconds % 60
if hours > 0:
return f"{hours}:{minutes:02d}:{secs:02d}"
else:
return f"{minutes}:{secs:02d}"
def format_pace(pace_min_per_mile: Optional[float]) -> str:
"""Format pace in minutes per mile to MM:SS/mi"""
if pace_min_per_mile is None:
return "--:--/mi"
minutes = int(pace_min_per_mile)
seconds = int((pace_min_per_mile - minutes) * 60)
return f"{minutes}:{seconds:02d}/mi"
def format_speed(speed_mph: Optional[float]) -> str:
"""Format speed to mph with one decimal place"""
if speed_mph is None:
return "-- mph"
return f"{speed_mph:.1f} mph"
def format_distance(distance_miles: Optional[float]) -> str:
"""Format distance to miles with two decimal places"""
if distance_miles is None:
return "-- mi"
return f"{distance_miles:.2f} mi"
def format_workout_date(timestamp: Optional[str]) -> str:
"""Format workout timestamp to human-readable date"""
if timestamp is None:
return "Unknown date"
try:
dt = datetime.fromisoformat(timestamp.replace('Z', '+00:00'))
return dt.strftime("%b %d, %Y %I:%M %p")
except (ValueError, AttributeError):
return "Invalid date"
def format_workout_date_short(timestamp: Optional[str]) -> str:
"""Format workout timestamp for tabular display (shorter format)"""
if timestamp is None:
return "Unknown"
try:
dt = datetime.fromisoformat(timestamp.replace('Z', '+00:00'))
return dt.strftime("%m/%d/%Y %H:%M")
except (ValueError, AttributeError):
return "Invalid"
def format_activity_timestamp(timestamp: Optional[str]) -> str:
"""Format activity file timestamp for tabular display"""
if timestamp is None:
return "Unknown"
try:
# Handle both ISO formats with and without timezone info
if timestamp.endswith('Z'):
dt = datetime.fromisoformat(timestamp.replace('Z', '+00:00'))
elif '+' in timestamp or '-' in timestamp[-6:]:
# Handle timezone offset like -05:00
dt = datetime.fromisoformat(timestamp)
else:
dt = datetime.fromisoformat(timestamp)
return dt.strftime("%Y-%m-%d %H:%M:%S")
except (ValueError, AttributeError):
return "Invalid date"
def format_workout_summary(workout: Workout) -> str:
"""Create a one-line summary of a workout"""
distance = format_distance(workout.distance_miles)
duration = format_time_duration(workout.time_seconds)
pace = format_pace(workout.pace_min_per_mile)
parts = []
if workout.distance_miles:
parts.append(distance)
if workout.time_seconds:
parts.append(f"in {duration}")
if workout.pace_min_per_mile:
parts.append(f"({pace} pace)")
if parts:
return " ".join(parts)
else:
return "Workout data"
def estimate_calories(distance_miles: Optional[float], time_seconds: Optional[Union[int, float]], user_weight: float = 150) -> Optional[int]:
"""Estimate calories burned based on distance and time"""
if not distance_miles or not time_seconds:
return None
pace = (time_seconds / 60) / distance_miles
met = 8.0
if pace < 6:
met = 15.0
elif pace < 7:
met = 12.0
elif pace < 8:
met = 10.0
elif pace < 9:
met = 8.5
elif pace < 10:
met = 8.0
else:
met = 7.0
return int(met * (user_weight / 2.2) * (time_seconds / 3600))
# Column widths for workout table
_COL_ID = 3
_COL_DATE = 16
_COL_DISTANCE = 10
_COL_DURATION = 9
_COL_PACE = 10
_COL_SPEED = 9
_COL_SOURCE = 7
def print_workout_table_header():
"""Print the header for a workout table"""
print(f"{'ID':<{_COL_ID}} {'Date':<{_COL_DATE}} {'Distance':<{_COL_DISTANCE}} {'Duration':<{_COL_DURATION}} {'Pace':<{_COL_PACE}} {'Speed':<{_COL_SPEED}} {'Source':<{_COL_SOURCE}}")
print("-" * (_COL_ID + _COL_DATE + _COL_DISTANCE + _COL_DURATION + _COL_PACE + _COL_SPEED + _COL_SOURCE + 6)) # +6 for spaces between columns
def print_workout_row(workout: Workout):
"""Print a single workout as a table row"""
date = format_workout_date_short(workout.start_time or workout.created_at)
distance = format_distance(workout.distance_miles)
duration = format_time_duration(workout.time_seconds)
pace = format_pace(workout.pace_min_per_mile)
speed = format_speed(workout.speed_mph)
source = "FIT" if workout.activity_file_id else "Manual"
print(f"{workout.id:<{_COL_ID}} {date:<{_COL_DATE}} {distance:<{_COL_DISTANCE}} {duration:<{_COL_DURATION}} {pace:<{_COL_PACE}} {speed:<{_COL_SPEED}} {source:<{_COL_SOURCE}}")
def print_workout_details(workout: Workout):
"""Print detailed workout information"""
print(f"Workout #{workout.id}")
print(f" Date: {format_workout_date(workout.start_time or workout.created_at)}")
if workout.distance_miles:
print(f" Distance: {format_distance(workout.distance_miles)}")
if workout.time_seconds:
print(f" Duration: {format_time_duration(workout.time_seconds)}")
if workout.pace_min_per_mile:
print(f" Pace: {format_pace(workout.pace_min_per_mile)}")
if workout.speed_mph:
print(f" Speed: {format_speed(workout.speed_mph)}")
if workout.start_time and workout.end_time:
start = format_workout_date(workout.start_time)
end = format_workout_date(workout.end_time)
print(f" Start: {start}")
print(f" End: {end}")
estimated_calories = estimate_calories(workout.distance_miles, workout.time_seconds)
if estimated_calories:
print(f" Estimated Calories: {estimated_calories}")
source = "FIT file" if workout.activity_file_id else "Manual entry"
print(f" Source: {source}")
if workout.activity_file_id:
print(f" Activity File ID: {workout.activity_file_id}")
print(f" Created: {format_workout_date(workout.created_at)}")
print(f" Updated: {format_workout_date(workout.updated_at)}")

View File

@@ -1,42 +0,0 @@
import argparse
import pathlib
import logging
import trygo_py_cliclient.config
import trygo_py_cliclient.cli.common
import trygo_py_cliclient.cli.register
import trygo_py_cliclient.cli.login
_logger = logging.getLogger(__name__)
def parse_args():
parser = argparse.ArgumentParser(
"trygo-client", formatter_class=argparse.ArgumentDefaultsHelpFormatter
)
parser.add_argument(
"--config-file", type=str, help="config file location", default="config.toml"
)
subparsers = parser.add_subparsers(dest="cmd", required=True)
trygo_py_cliclient.cli.register.setup_parser(subparsers)
trygo_py_cliclient.cli.login.setup_parser(subparsers)
args = parser.parse_args()
return args
def main():
args = parse_args()
config = trygo_py_cliclient.config.read_config(pathlib.Path(args.config_file))
trygo_py_cliclient.cli.common.set_up_logging(
config,
)
_logger.info(f"Got args {args=}")
_logger.info(f"Loaded config {config=}")
# TODO is there a clean way to hang on to a session for a bit when the cli command has run?
# i guess realistically we don't want that
args.func(config, args)

View File

@@ -1,29 +0,0 @@
import argparse
import logging
import typing
import trygo_py_cliclient.config
import trygo_py_cliclient.client
_logger = logging.getLogger(__name__)
if typing.TYPE_CHECKING:
_SubparserType = argparse._SubParsersAction[argparse.ArgumentParser]
else:
_SubparserType = typing.Any
def setup_parser(subparsers: _SubparserType) -> None:
parser = subparsers.add_parser("login")
parser.add_argument("--email", type=str, help="The email to log in with ", default="")
parser.add_argument("--password", type=str, help="Password", default="")
parser.set_defaults(func=run)
def run(config: trygo_py_cliclient.config.Config, args):
clnt = trygo_py_cliclient.client.BackendService(config)
response = clnt.login(args.email, args.password)
_logger.info(response)

View File

@@ -1,149 +0,0 @@
import requests
import logging
from typing import Optional, Dict, Any
from dataclasses import asdict
from pathlib import Path
from trygo_py_cliclient.models import RegisterRequest, LoginRequest, AuthResponse
from trygo_py_cliclient.config import Config
from trygo_py_cliclient.exceptions import (
TryGoAPIError,
AuthenticationError,
ValidationError,
NetworkError,
ServerError,
)
_logger = logging.getLogger(__name__)
class BackendService:
def __init__(self, config: Config):
self.base_url = config.general_config.backend_base_url.rstrip("/")
self.timeout = 30
self.session = requests.Session()
self.session.headers.update(
{"Content-Type": "application/json", "User-Agent": "trygo-py-client/0.1.0"}
)
def _make_request(
self, method: str, endpoint: str, data: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""
Make HTTP request to API endpoint
Args:
method: HTTP method (GET, POST, PUT, DELETE)
endpoint: API endpoint path
data: Request payload data
Returns:
Response data as dictionary
Raises:
NetworkError: For connection/timeout errors
AuthenticationError: For 401 errors
ValidationError: For 400 errors
ServerError: For 5xx errors
TryGoAPIError: For other API errors
"""
url = f"{self.base_url}{endpoint}"
_logger.info(f"Making {method} request to {url}")
try:
response = self.session.request(
method=method, url=url, json=data, timeout=self.timeout
)
# Log response status
_logger.debug(f"Response status: {response.status_code}")
# Handle different HTTP status codes
if response.status_code == 401:
raise AuthenticationError("Authentication failed")
elif response.status_code == 400:
error_msg = "Validation error"
try:
error_data = response.json()
if "message" in error_data:
error_msg = error_data["message"]
except Exception:
pass
raise ValidationError(error_msg)
elif 500 <= response.status_code < 600:
raise ServerError(f"Server error: {response.status_code}")
elif not response.ok:
raise TryGoAPIError(f"API error: {response.status_code}")
# Parse JSON response
try:
return response.json()
except ValueError as e:
raise TryGoAPIError(f"Invalid JSON response: {e}")
# claude really fucked this
except requests.exceptions.Timeout:
raise NetworkError("Request timeout")
except requests.exceptions.ConnectionError:
raise NetworkError("Connection error")
except requests.exceptions.RequestException as e:
raise NetworkError(f"Network error: {e}")
def register(self, email: str, password: str, display_name: str) -> AuthResponse:
"""
Register a new user account
"""
request_data = RegisterRequest(
email=email, password=password, display_name=display_name
)
_logger.info(f"Registering user: {email}")
response_data = self._make_request(
"POST", "/auth/register", asdict(request_data)
)
# Parse response and create AuthResponse
auth_response = AuthResponse(
message=response_data.get("message", ""),
email=response_data.get("email", ""),
id=response_data.get("id", ""),
)
return auth_response
def login(self, email: str, password: str) -> str:
"""
Authenticate user and get auth token
"""
request_data = LoginRequest(email=email, password=password)
_logger.info(f"Logging in user: {email}")
response_data = self._make_request("POST", "/auth/tokens", asdict(request_data))
_logger.info(response_data)
# Parse response and create AuthResponse
token = response_data.get("token", "")
self._store_credential(token)
return token
def logout(self) -> None:
"""
Clear authentication token and session data
Optionally calls logout endpoint if server supports it
"""
_logger.info("Logging out user")
# Try to call logout endpoint if we have a token
_logger.debug("Local authentication cleared")
def _store_credential(self, token: str) -> None:
token_path = Path("cred/token")
token_path.parent.mkdir(parents=True,exist_ok=True)
with token_path.open(mode="w") as tokenf:
tokenf.write(token)

View File

@@ -1,10 +0,0 @@
from trygo_py_cliclient.config.config import (
GeneralConfig,
Config,
)
from trygo_py_cliclient.config.config_reader import (
read_config,
)
__all__ = ["GeneralConfig", "Config", "read_config"]

View File

@@ -1,38 +0,0 @@
from dataclasses import dataclass
from typing import Optional
@dataclass
class User:
"""User model for API requests and responses"""
email: str
password: str
display_name: str
id: Optional[str] = None
@dataclass
class RegisterRequest:
"""Request payload for /auth/register endpoint"""
email: str
password: str
display_name: str
@dataclass
class LoginRequest:
"""Request payload for /auth/login endpoint"""
email: str
password: str
@dataclass
class AuthResponse:
"""Response from authentication endpoints"""
message: str
email: Optional[str] = None
id: Optional[str] = None

14
uv.lock generated
View File

@@ -132,6 +132,7 @@ dependencies = [
{ name = "dacite" }, { name = "dacite" },
{ name = "requests" }, { name = "requests" },
{ name = "tomli" }, { name = "tomli" },
{ name = "types-requests" },
] ]
[package.dev-dependencies] [package.dev-dependencies]
@@ -149,6 +150,7 @@ requires-dist = [
{ name = "dacite", specifier = ">=1.9.2" }, { name = "dacite", specifier = ">=1.9.2" },
{ name = "requests", specifier = ">=2.31.0" }, { name = "requests", specifier = ">=2.31.0" },
{ name = "tomli", specifier = ">=2.2.1" }, { name = "tomli", specifier = ">=2.2.1" },
{ name = "types-requests", specifier = ">=2.32.4.20250913" },
] ]
[package.metadata.requires-dev] [package.metadata.requires-dev]
@@ -367,6 +369,18 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/6e/c2/61d3e0f47e2b74ef40a68b9e6ad5984f6241a942f7cd3bbfbdbd03861ea9/tomli-2.2.1-py3-none-any.whl", hash = "sha256:cb55c73c5f4408779d0cf3eef9f762b9c9f147a77de7b258bef0a5628adc85cc", size = 14257, upload-time = "2024-11-27T22:38:35.385Z" }, { url = "https://files.pythonhosted.org/packages/6e/c2/61d3e0f47e2b74ef40a68b9e6ad5984f6241a942f7cd3bbfbdbd03861ea9/tomli-2.2.1-py3-none-any.whl", hash = "sha256:cb55c73c5f4408779d0cf3eef9f762b9c9f147a77de7b258bef0a5628adc85cc", size = 14257, upload-time = "2024-11-27T22:38:35.385Z" },
] ]
[[package]]
name = "types-requests"
version = "2.32.4.20250913"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "urllib3" },
]
sdist = { url = "https://files.pythonhosted.org/packages/36/27/489922f4505975b11de2b5ad07b4fe1dca0bca9be81a703f26c5f3acfce5/types_requests-2.32.4.20250913.tar.gz", hash = "sha256:abd6d4f9ce3a9383f269775a9835a4c24e5cd6b9f647d64f88aa4613c33def5d", size = 23113, upload-time = "2025-09-13T02:40:02.309Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/2a/20/9a227ea57c1285986c4cf78400d0a91615d25b24e257fd9e2969606bdfae/types_requests-2.32.4.20250913-py3-none-any.whl", hash = "sha256:78c9c1fffebbe0fa487a418e0fa5252017e9c60d1a2da394077f1780f655d7e1", size = 20658, upload-time = "2025-09-13T02:40:01.115Z" },
]
[[package]] [[package]]
name = "typing-extensions" name = "typing-extensions"
version = "4.12.2" version = "4.12.2"