Example using the farmOS.py classes in jupyterlite.

jupyterlite_farmospy.ipynb

{
  "metadata": {
    "language_info": {
      "codemirror_mode": {
        "name": "python",
        "version": 3
      },
      "file_extension": ".py",
      "mimetype": "text/x-python",
      "name": "python",
      "nbconvert_exporter": "python",
      "pygments_lexer": "ipython3",
      "version": "3.8"
    },
    "kernelspec": {
      "name": "python",
      "display_name": "Pyolite",
      "language": "python"
    }
  },
  "nbformat_minor": 4,
  "nbformat": 4,
  "cells": [
    {
      "cell_type": "code",
      "source": "import json\nimport micropip\nfrom pyodide import to_js\nfrom IPython.display import JSON\nfrom js import Object, fetch\n\n# farmOS.py\nmicropip.install(['farmos==1.0.0b3'])\nfrom farmOS.client_2 import ResourceBase\n\n# Create farmOS demo at https://laughing-mclean-bcdbff.netlify.app/\n# Configure correct cors for farm client.\nurl=\"https://main-6npvoy6ca1czylqdtjxi9gcednd3eclb.tugboat.qa\"\n\n# Create a response class.\n# farmOS.py internals call response.json() so we need to provide this method.\n# farmOS.py also references response.status_code.\n# Maybe use the Response class from requests?\nclass Response():\n    def __init__(self, data):\n        self.data = data\n        self.status_code = ''\n    def json(self):\n        return self.data\n\n# Create a custom JupyterSession that uses js.fetch.\nclass JupyterSession():\n    def __init__(self, hostname):\n        self.hostname = hostname\n        self.token = ''\n        self._content_type = \"application/vnd.api+json\"\n    \n    # Copy most of the authorize method from farmOS.py session.\n    # Use js.fetch to get access_token.\n    # TODO: support refresh token, etc, basically recreate requests_oauthlib\n    async def authorize(self, username=None, password=None, scope=None):\n        \"\"\"Authorize with the farmOS OAuth server.\"\"\"\n\n        # Ask for username if not provided.\n        if username is None:\n            from getpass import getpass\n\n            username = getpass(\"Enter username: \")\n\n        # Ask for password if not provided.\n        if password is None:\n            from getpass import getpass\n\n            password = getpass(\"Enter password: \")\n\n        # Use default scope if none provided.\n        if scope is None:\n            scope = \"farm_manager\"\n            \n        # Request a new token.\n        token_url= self.hostname + \"/oauth/token\"\n        full_body = f\"grant_type=password&username={username}&password={password}&client_id=farm&scope={scope}\"\n        resp = await fetch(token_url,\n          method=\"POST\",\n          body=full_body,  \n          headers=Object.fromEntries(to_js({ \"Content-Type\": \"application/x-www-form-urlencoded\" })),\n        )\n        res = await resp.text()\n        rj=json.loads(res)\n        token=rj['access_token']\n        \n        self.token = token\n        return token\n\n    async def http_request(self, path, method=\"GET\", options=None, params=None, headers=None):\n        # Strip protocol, hostname, leading/trailing slashes, and whitespace from the path.\n        path = path.strip(\"/\")\n        path = path.strip()\n\n        return await self._http_request(\n            url=path, method=method, options=options, params=params, headers=headers\n        )\n\n    async def _http_request(self, url, method=\"GET\", options=None, params=None, headers=None):\n\n        url = f\"{self.hostname}/{url}\"\n\n        if params is None:\n            params = {}\n\n        if headers is None:\n            headers = {}\n\n        # If there is a json data to be sent, include it.\n        json_data = None\n        if options and \"json\" in options:\n            # Convert data to json string.\n            json_data = json.dumps(options[\"json\"])\n            if \"Content-Type\" not in headers:\n                headers[\"Content-Type\"] = self._content_type\n\n        # Set authorization header.\n        headers[\"Authorization\"] = f\"Bearer {self.token}\"        \n        \n        # Perform the request.\n        # TODO support query params.\n        resp = await fetch(url,\n          method=method,\n          headers=Object.fromEntries(to_js(headers)),\n          # params = ???\n          body=json_data\n        )\n        res = await resp.text()\n        response = Response(json.loads(res))\n        return response\n\n# Extend the core ResourceBase class base get, send and delete methods to be async.\nclass JResourceBase(ResourceBase):\n    \"\"\"Base class for JSONAPI resource methods.\"\"\"\n\n    def __init__(self, session):\n        super().__init__(session)\n\n    async def _get_records(self, entity_type, bundle=None, resource_id=None, params=None):\n        \"\"\"Helper function that checks to retrieve one record, one page or multiple pages of farmOS records\"\"\"\n        if params is None:\n            params = {}\n\n        params = {**self.params, **params}\n\n        path = self._get_resource_path(entity_type, bundle, resource_id)\n\n        response = await self.session.http_request(path=path, params=params)\n        return response.json()\n    \n    async def send(self, entity_type, bundle=None, payload=None):\n\n        # Default to empty payload dict.\n        if payload is None:\n            payload = {}\n\n        # Set the resource type.\n        payload[\"type\"] = self._get_resource_type(entity_type, bundle)\n        json_payload = {\n            \"data\": {**payload},\n        }\n        options = {\"json\": json_payload}\n\n        # If an ID is included, update the record\n        id = payload.pop(\"id\", None)\n        if id:\n            json_payload[\"data\"][\"id\"] = id\n            path = self._get_resource_path(\n                entity_type=entity_type, bundle=bundle, record_id=id\n            )\n            response = await self.session.http_request(\n                method=\"PATCH\", path=path, options=options\n            )\n        # If no ID is included, create a new record\n        else:\n            path = self._get_resource_path(entity_type=entity_type, bundle=bundle)\n            response = await self.session.http_request(\n                method=\"POST\", path=path, options=options\n            )\n            \n        return response.json()\n\n    async def delete(self, entity_type, bundle=None, id=None):\n        path = self._get_resource_path(\n            entity_type=entity_type, bundle=bundle, record_id=id\n        )\n        return await self.session.http_request(method=\"DELETE\", path=path)\n\n# Finally, authorize and instantiate the resource base class.\nsession = JupyterSession(url)\nawait session.authorize(\"manager\", \"manager\")\nresources = JResourceBase(session)",
      "metadata": {
        "trusted": true
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": "land = await resources.get('asset', 'land')\nlen(land[\"data\"])",
      "metadata": {
        "trusted": true
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": "new_asset = {\n    \"attributes\": {\n        \"name\": \"Test Land\",\n        \"status\": \"active\",\n        \"land_type\": \"field\",\n    }\n}\nawait resources.send('asset', 'land', new_asset)",
      "metadata": {
        "trusted": true
      },
      "execution_count": null,
      "outputs": []
    }
  ]
}