import os
from datetime import date, datetime
from typing import Any, Literal
import httpx
import pydantic
import pydantic.alias_generators
from dotenv import load_dotenv
load_dotenv()
API_TOKEN = os.getenv("MARVIN_API_TOKEN", "<secret>")
FULL_ACCESS_TOKEN = os.getenv("MARVIN_FULL_ACCESS_TOKEN", "<secret>")
def _get_timezone_offset_in_minutes() -> int:
offset = datetime.now().astimezone().utcoffset()
if offset is None:
raise RuntimeError("Could not determine timezone offset")
return int(offset.total_seconds() / 60)
class MarvinItem(pydantic.BaseModel):
model_config = pydantic.ConfigDict(
alias_generator=pydantic.alias_generators.to_camel, serialize_by_alias=True
)
done: bool = False
day: date | None = None
parent_id: None | str = None # ID of parent category/project
label_ids: None | list[str] = None # IDs of labels
first_scheduled: None | date = None
rank: None | int = None
daily_section: None | str = None
bonus_section: None | str = None
# ID of custom section (from profile.strategySettings.customStructure)
custom_section: None | str = None
time_block_section: None | str = None # ID of time block
note: None | str = None
due_date: None | date = None
time_estimate: None | int = None # milliseconds
is_reward: None | bool = None
is_frogged: None | int = None
planned_week: None | date = None
planned_month: None | str = None
reward_points: None | float = None
reward_id: None | str = None # Unique ID of attached Reward
# Manually put in backburner (can also be backburner from label, start date, etc.)
backburner: None | bool = None
review_date: None | date = None
item_snooze_time: None | int = None # Date.now() until when it's snoozed
perma_snooze_time: None | str = None
# Added to time to fix time zone issues. So if the user is in Pacific time,
# this would be -8*60. If the user added a task with +today at 22:00 local
# time 2019-12-05, then we want to create the task on 2019-12-05 and not
# 2019-12-06 (which is the server date).
timezone_offset: None | int = pydantic.Field(
default_factory=_get_timezone_offset_in_minutes
)
class MarvinTask(MarvinItem):
# supports some autocompletion (parent, day, dueDate, plannedWeek, plannedMonth,
# timeEstimate, labels, isStarred).
# Use header X-Auto-Complete=false to disable autocompletion.
title: str
is_starred: None | int = None
class MarvinProject(MarvinItem):
# supports some autocompletion (parent, dueDate, labels)
title: str
priority: None | Literal["high", "mid", "low"] = None
class MarvinEvent(pydantic.BaseModel):
title: str
note: str | None = None # Use markdown
length: int | None = None # milliseconds
start: datetime | None = None # ISO formatted start time
class MarvinServer:
def __init__(self, base_url: str = "https://serv.amazingmarvin.com"):
"""Get info using Amazing Marvin API."""
self.base_url = base_url
def test_credentials(self) -> None:
"""Test whether your credentials are correct."""
response = httpx.post(
self.base_url + "/api/test", headers={"X-API-Token": API_TOKEN}
)
response.raise_for_status()
class MarvinServerWrite(MarvinServer):
def create_task(
self, task: MarvinTask, *, auto_complete: bool = True
) -> dict[str, Any]:
"""Create a task.
Unless you set auto_complete=False header, the title will be processed similar
to autocompletion within Marvin. So if you use "Example task +today", then the
task will be scheduled for today and " +today" will be removed from the title.
If you put "#Parent Category" or "@label1 @label2" in the title, they will be
resolved to their IDs when you open Marvin.
"""
headers = {"X-API-Token": API_TOKEN}
if not auto_complete:
headers["X-Auto-Complete"] = "false"
response = httpx.post(
self.base_url + "/api/addTask", headers=headers, json=task.model_dump_json()
)
response.raise_for_status()
return response.json() # type: ignore[no-any-return]
def mark_task_done(
self, task_id: str, timezone_offset: int | None = None
) -> dict[str, Any]:
"""Mark a task done.
timezone_offset, in minutes, is used to determine the date when the task was
marked done.
Many things happen in the Marvin client when you mark a task done.
Some of these are implemented, and more will be implemented in the future.
If a feature is missing that affects your workflow, please create an issue.
"""
if timezone_offset is None:
timezone_offset = _get_timezone_offset_in_minutes()
response = httpx.post(
self.base_url + "/api/markDone",
headers={"X-API-Token": API_TOKEN},
json={"itemId": task_id, "timeZoneOffset": timezone_offset},
)
response.raise_for_status()
return response.json() # type: ignore[no-any-return]
def create_project(self, project: MarvinProject) -> dict[str, Any]:
"""Create a project."""
response = httpx.post(
self.base_url + "/api/addProject",
headers={"X-API-Token": API_TOKEN},
json=project.model_dump_json(),
)
response.raise_for_status()
return response.json() # type: ignore[no-any-return]
def create_event(self, event: MarvinEvent) -> dict[str, Any]:
"""Create an event."""
response = httpx.post(
self.base_url + "/api/addEvent",
headers={"X-API-Token": API_TOKEN},
json=event.model_dump_json(),
)
response.raise_for_status()
return response.json() # type: ignore[no-any-return]
class MarvinServerRead(MarvinServer):
def _get_from_api(
self,
endpoint: str,
params: dict[str, str | list[str]] | None = None,
*,
full_access_token: bool = False,
) -> Any: # noqa: ANN401
headers = {}
if full_access_token:
headers["X-Full-Access-Token"] = FULL_ACCESS_TOKEN
else:
headers["X-API-Token"] = API_TOKEN
response = httpx.get(self.base_url + endpoint, headers=headers, params=params)
response.raise_for_status()
if response.status_code == httpx.codes.NO_CONTENT:
return None
return response.json()
def get_currently_tracked_task(self) -> dict[str, Any]:
"""Get the currently tracked task.
=>
{
"_id": "a12345",
"db": "Tasks",
"title": "Work 30m on homework"
}
"""
return self._get_from_api("/api/trackedItem") # type: ignore[no-any-return]
def get_child_items(self, parent_id: str = "unassigned") -> list[dict[str, Any]]:
"""Get child tasks/projects of a category/project.
Marvin has infinite nesting of tasks/projects with categories/projects.
The ID of an item's parent is stored in its parentId field.
This returns all items that have parentId equal to the ID you provide.
You will have to do additional queries if you want to retrieve the nested
grandchildren, and further descendants.
Note: this endpoint only returns open items.
=>
[...]
The response is an array of Projects and Tasks.
"""
return self._get_from_api("/api/children", params={"parentId": parent_id}) # type: ignore[no-any-return]
def get_items_scheduled_today(
self, today: date | None = None
) -> list[dict[str, Any]]:
"""Get tasks and projects scheduled today.
Including rollover/auto-schedule due items if enabled.
=>
[...]
"""
if today is None:
today = date.today() # noqa: DTZ011
return self._get_from_api("/api/todayItems", params={"date": today.isoformat()}) # type: ignore[no-any-return]
def get_open_items_due_today(self, by: date | None = None) -> list[dict[str, Any]]:
"""Get open Tasks and Projects that are due today (or earlier).
=>
[...]
"""
if by is None:
by = date.today() # noqa: DTZ011
return self._get_from_api("/api/dueItems", params={"by": by.isoformat()}) # type: ignore[no-any-return]
def get_today_time_blocks(self, today: date | None = None) -> list[dict[str, Any]]:
"""Get a list of today's time blocks.
=>
[...]
The response is an array of Time Blocks.
"""
if today is None:
today = date.today() # noqa: DTZ011
return self._get_from_api( # type: ignore[no-any-return]
"/api/todayTimeBlocks", params={"date": today.isoformat()}
)
def get_categories(self) -> list[dict[str, Any]]:
"""Get a list of all categories.
You can access a list of all your categories with a GET request.
See Categories for documentation of the data format.
=>
[{ "_id": "a4123412", "title": "Work", "parentId": "root", "color": "#4184a0",
... }, ...]
"""
return self._get_from_api("/api/categories") # type: ignore[no-any-return]
def get_labels(self) -> list[dict[str, Any]]:
"""Get a list of all labels, in their sort order (used in sort by label).
The label groups can be found by reading the sync database document
profile.strategySettings.labelSettings.groups.
See documentation for labels and label groups
=>
[{ "_id": "abcdefg1234", "title": "quick", "color": "#f0a0a0", "icon": "tag",
"groupId": "5421fabcdae" }, ...]
"""
return self._get_from_api("/api/labels") # type: ignore[no-any-return]
def get_time_track_info(self, task_ids: list[str]) -> list[dict[str, Any]]:
"""Get time track info.
Marvin caches time track data in task.times and task.duration.
But if you use the API to do time tracking and don't update the task,
or if something went wrong with sync, then this info might be missing.
You can get the source of truth about time track data using the /api/tracks
endpoint.
You can request the time track data for up to 100 tasks at once.
This API will return an array of time track data with taskId and a times array.
See Marvin data types#Tasks for documentation about times.
=>
[
{ "taskId": "nfieamewpemnfiqe", "times": [1625985037144, 1625985043470,
1625985049000, 1625985054374] }
]
"""
return self._get_from_api("/api/tracks", params={"taskIds": task_ids}) # type: ignore[no-any-return]
def get_kudos_info(self) -> dict[str, Any]:
"""Get Marvin Kudos info.
=>
{
"kudos": 0,
"level": 1,
"kudosRemaining": 350
}
"""
return self._get_from_api("/api/kudos") # type: ignore[no-any-return]
def get_account_info(self) -> dict[str, Any]:
"""Retrieve some information about your account.
See Marvin Data Types#Profile for documentation on which fields are available
and what they mean.
=>
{
"email": "test@amazingmarvin.com",
...
}
"""
return self._get_from_api("/api/me") # type: ignore[no-any-return]
def get_reminders(self) -> list[dict[str, Any]]:
"""Get a list of all reminders that are currently scheduled.
See documentation for Reminders.
=>
Reminder[]
"""
return self._get_from_api("/api/reminders", full_access_token=True) or []
def get_goals(self) -> list[dict[str, Any]]:
"""Get Goals.
See Goals for documentation of the data format.
=>
[{ _id: "123", title: "Example Goal", ... }]
"""
return self._get_from_api("/api/goals") # type: ignore[no-any-return]
def get_habit(self, habit_id: str) -> dict[str, Any]:
"""Get a Habit, including its full history.
=>
{...}
"""
return self._get_from_api("/api/habit", params={"id": habit_id}) # type: ignore[no-any-return]
def get_habits(self, *, raw: bool = False) -> list[dict[str, Any]]:
"""Retrieve a list of all your Habits, with their full history.
With raw=True, the entire habit objects will be returned,
rather than just history information.
=>
[...]
"""
return self._get_from_api( # type: ignore[no-any-return]
"/api/habits", params={"raw": "1"} if raw else None, full_access_token=raw
)
def main() -> None:
"""Test the API."""
server = MarvinServerRead()
for attr in dir(server):
if attr.startswith("get_"):
print(attr) # noqa: T201
try:
print(getattr(server, attr)()) # noqa: T201
except Exception as e: # noqa: BLE001
print(e) # noqa: T201
print() # noqa: T201
if __name__ == "__main__":
main()
