Identifying headers missing from SDK requests
The 'User Agent' and 'X-Planet-App' headers are used to self-identify the SDK when communicating with the Planet services. These headers are not being sent. Fix the SDK logic so they are indeed being fixed.

example:
```console
$> planet orders request --id 20221103_091114_08_2451 --name test PSScene visual | planet --verbosity debug orders create

2022-11-03 14:02:18,244 - asyncio - DEBUG - Using selector: KqueueSelector
2022-11-03 14:02:18,244 - planet.auth - DEBUG - Reading from /Users/jennifer.kyle/.planet.json
2022-11-03 14:02:18,245 - planet.auth - DEBUG - Auth read from secret file /Users/jennifer.kyle/.planet.json.
2022-11-03 14:02:18,245 - planet.http - INFO - Session read timeout set to 30.0.
2022-11-03 14:02:18,257 - planet.http - DEBUG - Throttling cadence set to 0.1s.
2022-11-03 14:02:18,257 - planet.http - DEBUG - Workers capped at 50.
2022-11-03 14:02:18,258 - planet.http - DEBUG - Worker acquired.
2022-11-03 14:02:18,258 - planet.http - INFO - POST https://api.planet.com/compute/ops/orders/v2 - Sent
2022-11-03 14:02:18,258 - planet.http - DEBUG - POST https://api.planet.com/compute/ops/orders/v2 - Headers({'host': 'api.planet.com', 'content-type': 'application/json', 'content-length': '151', 'authorization': '[secure]'})
2022-11-03 14:02:18,258 - planet.http - DEBUG - POST https://api.planet.com/compute/ops/orders/v2 - b'{"name": "test", "products": [{"item_ids": ["20221103_091114_08_2451"], "item_type": "PSScene", "product_bundle": "visual"}], "metadata": {"stac": {}}}'
2022-11-03 14:02:19,047 - httpx._client - DEBUG - HTTP Request: POST https://api.planet.com/compute/ops/orders/v2 "HTTP/1.1 202 Accepted"
2022-11-03 14:02:19,047 - planet.http - INFO - POST https://api.planet.com/compute/ops/orders/v2 - Status 202
2022-11-03 14:02:19,047 - planet.http - DEBUG - Headers({'content-type': 'application/json', 'date': 'Thu, 03 Nov 2022 21:02:19 GMT', 'strict-transport-security': 'max-age=15724800; includeSubDomains', 'vary': 'Origin', 'x-planet-span-id': '9fefdcee-659c-4b03-9373-984a612aa32f', 'x-planet-trace-id': 'e4985228-f779-47ae-bb7e-b69d92e52576', 'x-request-id': 'e4985228-f779-47ae-bb7e-b69d92e52576', 'content-length': '438', 'via': '1.1 edge, 1.1 google', 'alt-svc': 'h3=":443"; ma=2592000,h3-29=":443"; ma=2592000'})
2022-11-03 14:02:19,048 - planet.http - DEBUG - Worker released.
{"_links": {"_self": "https://api.planet.com/compute/ops/orders/v2/bfcc1c34-8b3b-4e91-addf-dfef8fc90b5f"}, "created_on": "2022-11-03T21:02:18.673Z", "error_hints": [], "id": "bfcc1c34-8b3b-4e91-addf-dfef8fc90b5f", "last_message": "Preparing order", "last_modified": "2022-11-03T21:02:18.673Z", "metadata": {"stac": {}}, "name": "test", "products": [{"item_ids": ["20221103_091114_08_2451"], "item_type": "PSScene", "product_bundle": "visual"}], "state": "queued"}
```
In the output, the following line should have the 'User Agent' and 'X-Planet-App' headers but they are missing:
```
2022-11-03 14:02:18,258 - planet.http - DEBUG - POST https://api.planet.com/compute/ops/orders/v2 - Headers({'host': 'api.planet.com', 'content-type': 'application/json', 'content-length': '151', 'authorization': '[secure]'})
```
The session-level headers are getting set when the Session is created, but they are not merged into the request-level headers when a request is made. This merging happens in httpx with [build_request()](https://github.com/encode/httpx/blob/1aea9539bbe93b26103e3a722ba0c421f7eb7f82/httpx/_client.py#L322), but we are bypassing that function and going directly to calling send(). So we do not get the merging behavior httpx provides.

Two options:
1. move up a level from calling send() directly so as to use httpx's merging logic *(may take some work figuring out why we went directly to send())*
1. merge headers manually in Session._request() *(easier but wanders farther off the usage path httpx anticipates)*




diff --git a/planet/auth.py b/planet/auth.py
index a7b5d51..5a64ef3 100644
--- a/planet/auth.py
+++ b/planet/auth.py
@@ -24,7 +24,7 @@ import typing
 import httpx
 import jwt
 
-from . import http, models
+from . import http
 from .constants import PLANET_BASE_URL, SECRET_FILE_PATH
 from .exceptions import AuthException
 from typing import Optional
@@ -177,10 +177,8 @@ class AuthClient:
         data = {'email': email, 'password': password}
 
         sess = http.AuthSession()
-        req = models.Request(url, method='POST', json=data)
-        resp = sess.request(req)
-        auth_data = self.decode_response(resp)
-        return auth_data
+        resp = sess.request(url=url, method='POST', json=data)
+        return self.decode_response(resp)
 
     @staticmethod
     def decode_response(response):
diff --git a/planet/clients/data.py b/planet/clients/data.py
index a94576b..fdba667 100644
--- a/planet/clients/data.py
+++ b/planet/clients/data.py
@@ -22,7 +22,7 @@ from typing import Any, AsyncIterator, Callable, Dict, List, Optional
 from .. import exceptions
 from ..constants import PLANET_BASE_URL
 from ..http import Session
-from ..models import Paged, Request, Response, StreamingBody
+from ..models import Paged, StreamingBody
 
 BASE_URL = f'{PLANET_BASE_URL}/data/v1/'
 SEARCHES_PATH = '/searches'
@@ -93,17 +93,6 @@ class DataClient:
     def _item_url(self, item_type, item_id):
         return f'{self._base_url}/item-types/{item_type}/items/{item_id}'
 
-    def _request(self, url, method, data=None, params=None, json=None):
-        return Request(url, method=method, data=data, params=params, json=json)
-
-    async def _do_request(self, request: Request) -> Response:
-        """Submit a request and get response.
-
-        Parameters:
-            request: request to submit
-        """
-        return await self._session.request(request)
-
     async def search(self,
                      item_types: List[str],
                      search_filter: dict,
@@ -185,11 +174,11 @@ class DataClient:
                     f'{sort} must be one of {SEARCH_SORT}')
             params['_sort'] = sort
 
-        request = self._request(url,
-                                method='POST',
-                                json=request_json,
-                                params=params)
-        return Items(request, self._do_request, limit=limit)
+        response = await self._session.request(method='POST',
+                                               url=url,
+                                               json=request_json,
+                                               params=params)
+        return Items(response, self._session.request, limit=limit)
 
     async def create_search(self,
                             name: str,
@@ -229,15 +218,16 @@ class DataClient:
         url = self._searches_url()
 
         # TODO: validate item_types
-        request_json = {
+        request = {
             'name': name,
             'filter': search_filter,
             'item_types': item_types,
             '__daily_email_enabled': enable_email
         }
 
-        request = self._request(url, method='POST', json=request_json)
-        response = await self._do_request(request)
+        response = await self._session.request(method='POST',
+                                               url=url,
+                                               json=request)
         return response.json()
 
     async def update_search(self,
@@ -260,15 +250,16 @@ class DataClient:
         """
         url = f'{self._searches_url()}/{search_id}'
 
-        request_json = {
+        request = {
             'name': name,
             'filter': search_filter,
             'item_types': item_types,
             '__daily_email_enabled': enable_email
         }
 
-        request = self._request(url, method='PUT', json=request_json)
-        response = await self._do_request(request)
+        response = await self._session.request(method='PUT',
+                                               url=url,
+                                               json=request)
         return response.json()
 
     async def list_searches(self,
@@ -307,8 +298,9 @@ class DataClient:
                 f'{search_type} must be one of {LIST_SEARCH_TYPE}')
 
         url = f'{self._searches_url()}'
-        request = self._request(url, method='GET')
-        return Searches(request, self._do_request, limit=limit)
+
+        response = await self._session.request(method='GET', url=url)
+        return Searches(response, self._session.request, limit=limit)
 
     async def delete_search(self, search_id: str):
         """Delete an existing saved search.
@@ -321,8 +313,7 @@ class DataClient:
         """
         url = f'{self._searches_url()}/{search_id}'
 
-        request = self._request(url, method='DELETE')
-        await self._do_request(request)
+        await self._session.request(method='DELETE', url=url)
 
     async def get_search(self, search_id: str) -> dict:
         """Get a saved search by id.
@@ -337,9 +328,9 @@ class DataClient:
             planet.exceptions.APIError: On API error.
         """
         url = f'{self._searches_url()}/{search_id}'
-        req = self._request(url, method='GET')
-        resp = await self._do_request(req)
-        return resp.json()
+
+        response = await self._session.request(method='GET', url=url)
+        return response.json()
 
     async def run_search(self,
                          search_id: str,
@@ -359,8 +350,8 @@ class DataClient:
         """
         url = f'{self._searches_url()}/{search_id}/results'
 
-        request = self._request(url, method='GET')
-        return Items(request, self._do_request, limit=limit)
+        response = await self._session.request(method='GET', url=url)
+        return Items(response, self._session.request, limit=limit)
 
     async def get_stats(self,
                         item_types: List[str],
@@ -388,14 +379,15 @@ class DataClient:
 
         url = f'{self._base_url}{STATS_PATH}'
 
-        request_json = {
+        request = {
             'interval': interval,
             'filter': search_filter,
             'item_types': item_types
         }
 
-        request = self._request(url, method='POST', json=request_json)
-        response = await self._do_request(request)
+        response = await self._session.request(method='POST',
+                                               url=url,
+                                               json=request)
         return response.json()
 
     async def list_asset_types(self) -> List[dict]:
@@ -500,8 +492,8 @@ class DataClient:
             planet.exceptions.APIError: On API error.
         """
         url = f'{self._item_url(item_type_id, item_id)}/assets'
-        request = self._request(url, method='GET')
-        response = await self._do_request(request)
+
+        response = await self._session.request(method='GET', url=url)
         return response.json()
 
     async def get_asset(self,
@@ -558,9 +550,8 @@ class DataClient:
 
         # lets not try to activate an asset already activating or active
         if status == 'inactive':
-            request = self._request(url, method='GET')
             # no response is returned
-            await self._do_request(request)
+            await self._session.request(method='GET', url=url)
 
         return
 
@@ -623,8 +614,7 @@ class DataClient:
                 raise exceptions.ClientError(
                     'asset missing ["_links"]["_self"] entry.')
 
-            request = self._request(asset_url, method='GET')
-            response = await self._do_request(request)
+            response = await self._session.request(method='GET', url=asset_url)
             asset = response.json()
 
         if max_attempts and num_attempts >= max_attempts:
@@ -669,9 +659,7 @@ class DataClient:
             raise exceptions.ClientError(
                 'asset missing ["location"] entry. Is asset active?')
 
-        req = self._request(location, method='GET')
-
-        async with self._session.stream(req) as resp:
+        async with self._session.stream(method='GET', url=location) as resp:
             body = StreamingBody(resp)
             dl_path = Path(directory, filename or body.name)
             dl_path.parent.mkdir(exist_ok=True, parents=True)
diff --git a/planet/clients/orders.py b/planet/clients/orders.py
index b37787d..e405046 100644
--- a/planet/clients/orders.py
+++ b/planet/clients/orders.py
@@ -25,7 +25,7 @@ from pathlib import Path
 from .. import exceptions
 from ..constants import PLANET_BASE_URL
 from ..http import Session
-from ..models import Paged, Request, Response, StreamingBody
+from ..models import Paged, StreamingBody
 
 BASE_URL = f'{PLANET_BASE_URL}/compute/ops'
 STATS_PATH = '/stats/orders/v2'
@@ -112,17 +112,6 @@ class OrdersClient:
     def _stats_url(self):
         return f'{self._base_url}{STATS_PATH}'
 
-    def _request(self, url, method, data=None, params=None, json=None):
-        return Request(url, method=method, data=data, params=params, json=json)
-
-    async def _do_request(self, request: Request) -> Response:
-        '''Submit a request and get response.
-
-        Parameters:
-            request: request to submit
-        '''
-        return await self._session.request(request)
-
     async def create_order(self, request: dict) -> dict:
         '''Create an order request.
 
@@ -156,10 +145,10 @@ class OrdersClient:
             planet.exceptions.APIError: On API error.
         '''
         url = self._orders_url()
-
-        req = self._request(url, method='POST', json=request)
-        resp = await self._do_request(req)
-        return resp.json()
+        response = await self._session.request(method='POST',
+                                               url=url,
+                                               json=request)
+        return response.json()
 
     async def get_order(self, order_id: str) -> dict:
         '''Get order details by Order ID.
@@ -177,9 +166,8 @@ class OrdersClient:
         self._check_order_id(order_id)
         url = f'{self._orders_url()}/{order_id}'
 
-        req = self._request(url, method='GET')
-        resp = await self._do_request(req)
-        return resp.json()
+        response = await self._session.request(method='GET', url=url)
+        return response.json()
 
     async def cancel_order(self, order_id: str) -> dict:
         '''Cancel a queued order.
@@ -197,9 +185,8 @@ class OrdersClient:
         self._check_order_id(order_id)
         url = f'{self._orders_url()}/{order_id}'
 
-        req = self._request(url, method='PUT')
-        resp = await self._do_request(req)
-        return resp.json()
+        response = await self._session.request(method='PUT', url=url)
+        return response.json()
 
     async def cancel_orders(self,
                             order_ids: Optional[List[str]] = None) -> dict:
@@ -224,9 +211,11 @@ class OrdersClient:
                 self._check_order_id(oid)
             cancel_body['order_ids'] = order_ids
 
-        req = self._request(url, method='POST', json=cancel_body)
-        resp = await self._do_request(req)
-        return resp.json()
+        response = await self._session.request(method='POST',
+                                               url=url,
+                                               json=cancel_body)
+
+        return response.json()
 
     async def aggregated_order_stats(self) -> dict:
         '''Get aggregated counts of active orders.
@@ -238,9 +227,8 @@ class OrdersClient:
             planet.exceptions.APIError: On API error.
         '''
         url = self._stats_url()
-        req = self._request(url, method='GET')
-        resp = await self._do_request(req)
-        return resp.json()
+        response = await self._session.request(method='GET', url=url)
+        return response.json()
 
     async def download_asset(self,
                              location: str,
@@ -264,9 +252,7 @@ class OrdersClient:
         Raises:
             planet.exceptions.APIError: On API error.
         """
-        req = self._request(location, method='GET')
-
-        async with self._session.stream(req) as resp:
+        async with self._session.stream(method='GET', url=location) as resp:
             body = StreamingBody(resp)
             dl_path = Path(directory, filename or body.name)
             dl_path.parent.mkdir(exist_ok=True, parents=True)
@@ -505,5 +491,7 @@ class OrdersClient:
         else:
             params = None
 
-        request = self._request(url, 'GET', params=params)
-        return Orders(request, self._do_request, limit=limit)
+        response = await self._session.request(method='GET',
+                                               url=url,
+                                               params=params)
+        return Orders(response, self._session.request, limit=limit)
diff --git a/planet/clients/subscriptions.py b/planet/clients/subscriptions.py
index a583a7c..888abfc 100644
--- a/planet/clients/subscriptions.py
+++ b/planet/clients/subscriptions.py
@@ -5,7 +5,7 @@ from typing import AsyncIterator, Optional, Set
 
 from planet.exceptions import APIError, ClientError
 from planet.http import Session
-from planet.models import Paged, Request
+from planet.models import Paged
 
 LOGGER = logging.getLogger()
 
@@ -29,7 +29,7 @@ class SubscriptionsClient:
     """
 
     def __init__(self, session: Session) -> None:
-        self.session = session
+        self._session = session
 
     async def list_subscriptions(self,
                                  status: Optional[Set[str]] = None,
@@ -63,11 +63,13 @@ class SubscriptionsClient:
 
         params = {'status': [val for val in status or {}]}
         url = 'https://api.planet.com/subscriptions/v1'
-        req = Request(url, params=params)
 
         try:
-            async for sub in _SubscriptionsPager(req,
-                                                 self.session.request,
+            response = await self._session.request(method='GET',
+                                                   url=url,
+                                                   params=params)
+            async for sub in _SubscriptionsPager(response,
+                                                 self._session.request,
                                                  limit=limit):
                 yield sub
         # Forward APIError. We don't strictly need this clause, but it
@@ -92,10 +94,11 @@ class SubscriptionsClient:
         """
 
         url = 'https://api.planet.com/subscriptions/v1'
-        req = Request(url, method='POST', json=request)
 
         try:
-            resp = await self.session.request(req)
+            resp = await self._session.request(method='POST',
+                                               url=url,
+                                               json=request)
         # Forward APIError. We don't strictly need this clause, but it
         # makes our intent clear.
         except APIError:
@@ -121,10 +124,9 @@ class SubscriptionsClient:
         """
         root_url = 'https://api.planet.com/subscriptions/v1'
         url = f'{root_url}/{subscription_id}/cancel'
-        req = Request(url, method='POST')
 
         try:
-            _ = await self.session.request(req)
+            _ = await self._session.request(method='POST', url=url)
         # Forward APIError. We don't strictly need this clause, but it
         # makes our intent clear.
         except APIError:
@@ -148,10 +150,11 @@ class SubscriptionsClient:
             ClientError: on a client error.
         """
         url = f'https://api.planet.com/subscriptions/v1/{subscription_id}'
-        req = Request(url, method='PUT', json=request)
 
         try:
-            resp = await self.session.request(req)
+            resp = await self._session.request(method='PUT',
+                                               url=url,
+                                               json=request)
         # Forward APIError. We don't strictly need this clause, but it
         # makes our intent clear.
         except APIError:
@@ -176,10 +179,9 @@ class SubscriptionsClient:
             ClientError: on a client error.
         """
         url = f'https://api.planet.com/subscriptions/v1/{subscription_id}'
-        req = Request(url, method='GET')
 
         try:
-            resp = await self.session.request(req)
+            resp = await self._session.request(method='GET', url=url)
         # Forward APIError. We don't strictly need this clause, but it
         # makes our intent clear.
         except APIError:
@@ -225,11 +227,14 @@ class SubscriptionsClient:
         params = {'status': [val for val in status or {}]}
         root_url = 'https://api.planet.com/subscriptions/v1'
         url = f'{root_url}/{subscription_id}/results'
-        req = Request(url, params=params)
 
         try:
-            async for sub in _ResultsPager(req,
-                                           self.session.request,
+            resp = await self._session.request(method='GET',
+                                               url=url,
+                                               params=params)
+
+            async for sub in _ResultsPager(resp,
+                                           self._session.request,
                                            limit=limit):
                 yield sub
         # Forward APIError. We don't strictly need this clause, but it
diff --git a/planet/http.py b/planet/http.py
index 4be3049..d554c4a 100644
--- a/planet/http.py
+++ b/planet/http.py
@@ -16,17 +16,17 @@
 from __future__ import annotations  # https://stackoverflow.com/a/33533514
 import asyncio
 from collections import Counter
+from contextlib import asynccontextmanager
 from http import HTTPStatus
 import logging
 import random
 import time
-
+from typing import AsyncGenerator, Optional
 import httpx
 
 from .auth import Auth, AuthType
 from . import exceptions, models
 from .__version__ import __version__
-from typing import Optional
 
 # NOTE: configuration of the session was performed using the data API quick
 # search endpoint. These values can be re-tested, tested with a new endpoint or
@@ -243,13 +243,16 @@ class Session(BaseSession):
 
         LOGGER.info(f'Session read timeout set to {READ_TIMEOUT}.')
         timeout = httpx.Timeout(10.0, read=READ_TIMEOUT)
+
+        headers = {
+            'User-Agent': self._get_user_agent(), 'X-Planet-App': 'python-sdk'
+        }
+
         self._client = httpx.AsyncClient(auth=auth,
+                                         headers=headers,
                                          timeout=timeout,
                                          follow_redirects=True)
 
-        self._client.headers.update({'User-Agent': self._get_user_agent()})
-        self._client.headers.update({'X-Planet-App': 'python-sdk'})
-
         async def alog_request(*args, **kwargs):
             return self._log_request(*args, **kwargs)
 
@@ -349,13 +352,17 @@ class Session(BaseSession):
         return min(calc_wait, max_retry_backoff)
 
     async def request(self,
-                      request: models.Request,
-                      stream: bool = False) -> models.Response:
-        """Submit a request with retry.
+                      method: str,
+                      url: str,
+                      json: Optional[dict] = None,
+                      params: Optional[dict] = None) -> models.Response:
+        """Build a request and submit it with retry and limiting.
 
         Parameters:
-            request: Request to submit.
-            stream: Get the body as a stream.
+            method: HTTP request method.
+            url: Location of the API endpoint.
+            json: JSON to send.
+            params: Values to send in the query string.
 
         Returns:
             Server response.
@@ -364,31 +371,47 @@ class Session(BaseSession):
             planet.exceptions.APIException: On API error.
             planet.exceptions.ClientError: When retry limit is exceeded.
         """
-        # TODO: retry will be provided in httpx v1 [1] with usage [2]
-        # 1. https://github.com/encode/httpcore/pull/221
-        # 2. https://github.com/encode/httpx/blob/
-        # 89fb0cbc69ea07b123dd7b36dc1ed9151c5d398f/docs/async.md#explicit-transport-instances # noqa
-        # TODO: if throttling is necessary, check out [1] once v1
-        # 1. https://github.com/encode/httpx/issues/984
-        return await self._retry(self._request, request, stream=stream)
+        if json:
+            headers = {'Content-Type': 'application/json'}
+        else:
+            headers = None
+
+        request = self._client.build_request(method=method,
+                                             url=url,
+                                             json=json,
+                                             params=params,
+                                             headers=headers)
+
+        http_response = await self._retry(self._send, request, stream=False)
+        return models.Response(http_response)
 
-    async def _request(self, request, stream=False):
-        """Submit a request with rate/worker limiting."""
+    async def _send(self, request, stream=False) -> httpx.Response:
+        """Send request with with rate/worker limiting."""
         async with self._limiter:
-            http_resp = await self._client.send(request.http_request,
-                                                stream=stream)
-        return models.Response(request, http_resp)
+            http_resp = await self._client.send(request, stream=stream)
 
-    def stream(self, request: models.Request) -> Stream:
+        return http_resp
+
+    @asynccontextmanager
+    async def stream(
+            self, method: str,
+            url: str) -> AsyncGenerator[models.StreamingResponse, None]:
         """Submit a request and get the response as a stream context manager.
 
         Parameters:
-            request: Request to submit
+            method: HTTP request method.
+            url: Location of the API endpoint.
 
         Returns:
-            Context manager providing the body as a stream.
+            Context manager providing the streaming response.
         """
-        return Stream(session=self, request=request)
+        request = self._client.build_request(method=method, url=url)
+        http_response = await self._retry(self._send, request, stream=True)
+        response = models.StreamingResponse(http_response)
+        try:
+            yield response
+        finally:
+            await response.aclose()
 
 
 class AuthSession(BaseSession):
@@ -404,11 +427,13 @@ class AuthSession(BaseSession):
             self._log_response, self._raise_for_status
         ]
 
-    def request(self, request):
+    def request(self, method: str, url: str, json: dict):
         """Submit a request
 
         Parameters:
-            request: Request to submit.
+            method: HTTP request method.
+            url: Location of the API endpoint.
+            json: JSON to send.
 
         Returns:
             Server response.
@@ -416,9 +441,9 @@ class AuthSession(BaseSession):
         Raises:
             planet.exceptions.APIException: On API error.
         """
-
-        http_resp = self._client.send(request.http_request)
-        return models.Response(request, http_resp)
+        request = self._client.build_request(method=method, url=url, json=json)
+        http_resp = self._client.send(request)
+        return models.Response(http_resp)
 
     @classmethod
     def _raise_for_status(cls, response):
@@ -428,26 +453,3 @@ class AuthSession(BaseSession):
             raise exceptions.APIError('Not a valid email address.')
         except exceptions.InvalidAPIKey:
             raise exceptions.APIError('Incorrect email or password.')
-
-
-class Stream:
-    '''Context manager for asynchronous response stream from Planet server.'''
-
-    def __init__(self, session: Session, request: models.Request):
-        """
-        Parameters:
-            session: Open session to Planet server.
-            request:  Request to submit.
-        """
-        self.session = session
-        self.request = request
-
-    async def __aenter__(self):
-        self.response = await self.session.request(
-            request=self.request,
-            stream=True,
-        )
-        return self.response
-
-    async def __aexit__(self, exc_type=None, exc_value=None, traceback=None):
-        await self.response.aclose()
diff --git a/planet/models.py b/planet/models.py
index 5bc79b5..ab38d8e 100644
--- a/planet/models.py
+++ b/planet/models.py
@@ -13,14 +13,13 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 """Manage data for requests and responses."""
-from datetime import datetime
 import logging
 import mimetypes
 from pathlib import Path
 import random
 import re
 import string
-import typing
+from typing import AsyncGenerator, Callable, List
 
 import httpx
 from tqdm.asyncio import tqdm
@@ -30,156 +29,87 @@ from .exceptions import PagingError
 LOGGER = logging.getLogger(__name__)
 
 
-class RequestException(Exception):
-    """Exceptions thrown by RequestException"""
-    pass
-
-
-class Request:
-    '''Handles a HTTP request for the Planet server.
-
-    :param url: URL of API endpoint
-    :type url: str
-    :param params: Values to send in the query string. Defaults to None.
-    :type params: dict, list of tuples, or bytes, optional
-    :param data: Object to send in the body. Defaults to None.
-    :type data: dict, list of tuples, bytes, or file-like object, optional
-    :param json: JSON to send. Defaults to None.
-    :type json: dict, optional
-    :param method: HTTP request method. Defaults to 'GET'
-    :type method: str, optional
-    :raises RequestException: When provided `body_type` is not a subclass of
-        :py:class:`planet.api.models.Body`
-    '''
-
-    def __init__(self, url, params=None, data=None, json=None, method='GET'):
-        if data or json:
-            headers = {'Content-Type': 'application/json'}
-        else:
-            headers = None
-
-        self.http_request = httpx.Request(method,
-                                          url,
-                                          params=params,
-                                          data=data,
-                                          json=json,
-                                          headers=headers)
+class Response:
+    """Handles the Planet server's response to a HTTP request."""
 
-    @property
-    def url(self):
-        return self.http_request.url
+    def __init__(self, http_response: httpx.Response):
+        """Initialize object.
 
-    @url.setter
-    def url(self, url):
-        '''Set the url.
+        Parameters:
+            http_response: Response that was received from the server.
+        """
+        self._http_response = http_response
 
-        :param url: URL of API endpoint
-        :type url: str
-        '''
-        self.http_request.url = httpx.URL(url)
+    def __repr__(self):
+        return f'<models.Response [{self.status_code}]>'
 
+    @property
+    def status_code(self) -> int:
+        """HTTP status code"""
+        return self._http_response.status_code
 
-class Response:
-    '''Handles the Planet server's response to a HTTP request.
+    def json(self) -> dict:
+        """Response json"""
+        return self._http_response.json()
 
-    :param request: Request that was submitted to the server
-    :type request: :py:Class:`Request`
-    :param http_response: Response that was received from the server
-    :type http_response: :py:Class:`requests.models.Response`
-    '''
 
-    def __init__(self, request, http_response):
-        self.request = request
-        self.http_response = http_response
-
-    def __repr__(self):
-        return f'<models.Response [{self.status_code}]>'
+class StreamingResponse(Response):
 
     @property
-    def status_code(self):
-        '''HTTP status code.
+    def headers(self) -> httpx.Headers:
+        return self._http_response.headers
 
-        :returns: status code
-        :rtype: int
-        '''
-        return self.http_response.status_code
+    @property
+    def url(self) -> str:
+        return str(self._http_response.url)
 
-    def json(self):
-        '''Get response json.
+    @property
+    def num_bytes_downloaded(self) -> int:
+        return self._http_response.num_bytes_downloaded
 
-        :returns:response json
-        :rtype: dict
-        '''
-        return self.http_response.json()
+    async def aiter_bytes(self):
+        async for c in self._http_response.aiter_bytes():
+            yield c
 
     async def aclose(self):
-        await self.http_response.aclose()
+        await self._http_response.aclose()
 
 
 class StreamingBody:
-    '''A representation of a streaming resource from the API.
+    """A representation of a streaming resource from the API."""
 
-    :param response: Response that was received from the server
-    :type response: :py:Class:`requests.models.Response`
-        '''
+    def __init__(self, response: StreamingResponse):
+        """Initialize the object.
 
-    def __init__(self, response):
-        self.response = response.http_response
-        self.url = response.request.url
+        Parameters:
+            response: Response that was received from the server.
+        """
+        self._response = response
 
     @property
-    def name(self):
-        '''The name of this resource.
+    def name(self) -> str:
+        """The name of this resource.
+
         The default is to use the content-disposition header value from the
         response. If not found, falls back to resolving the name from the url
         or generating a random name with the type from the response.
-
-        :returns: name of this resource
-        :rtype: str
-        '''
-        name = (_get_filename_from_headers(self.response.headers)
-                or _get_filename_from_url(self.url) or _get_random_filename(
-                    self.response.headers.get('content-type')))
+        """
+        name = (_get_filename_from_headers(self._response.headers)
+                or _get_filename_from_url(self._response.url)
+                or _get_random_filename(
+                    self._response.headers.get('content-type')))
         return name
 
     @property
-    def size(self):
-        '''The size of the body.
-
-        :returns: size of the body
-        :rtype: int
-        '''
-        return int(self.response.headers['Content-Length'])
-
-    @property
-    def num_bytes_downloaded(self):
-        '''The number of bytes downloaded.
-
-        :returns: number of bytes downloaded
-        :rtype: int
-        '''
-        return self.response.num_bytes_downloaded
-
-    def last_modified(self):
-        '''Read the last-modified header as a datetime, if present.
-
-        :returns: last-modified header
-        :rtype: datatime or None
-        '''
-        lm = self.response.headers.get('last-modified', None)
-        return datetime.strptime(lm, '%a, %d %b %Y %H:%M:%S GMT') if lm \\
-            else None
-
-    async def aiter_bytes(self):
-        async for c in self.response.aiter_bytes():
-            yield c
+    def size(self) -> int:
+        """The size of the body."""
+        return int(self._response.headers['Content-Length'])
 
     async def write(self,
                     filename: Path,
                     overwrite: bool = True,
                     progress_bar: bool = True):
         """Write the body to a file.
-
         Parameters:
             filename: Name to assign to downloaded file.
             overwrite: Overwrite any existing files.
@@ -221,10 +151,10 @@ class StreamingBody:
                           unit='B',
                           desc=str(filename),
                           disable=not progress_bar) as progress:
-                    previous = self.num_bytes_downloaded
-                    async for chunk in self.aiter_bytes():
+                    previous = self._response.num_bytes_downloaded
+                    async for chunk in self._response.aiter_bytes():
                         fp.write(chunk)
-                        new = self.num_bytes_downloaded
+                        new = self._response.num_bytes_downloaded
                         _log.update(new)
                         progress.update(new - previous)
                         previous = new
@@ -278,31 +208,26 @@ class Paged:
     ITEMS_KEY = 'items'
 
     def __init__(self,
-                 request: Request,
-                 do_request_fcn: typing.Callable,
+                 response: Response,
+                 request_fcn: Callable,
                  limit: int = 0):
         """
         Parameters:
             request: Request to send to server for first page.
-            do_request_fcn: Function for submitting a request. Must take in
-                Request object and return Response.
+            request_fcn: Function for submitting a request and retrieving a
+            result. Must take in url and method parameters.
             limit: Maximum number of results to return. When set to 0, no
                 maximum is applied.
         """
-        self.request = request
-        self._do_request = do_request_fcn
+        self._request_fcn = request_fcn
 
-        self._pages = self._get_pages()
+        self._pages = self._get_pages(response)
 
-        self._items: typing.List[dict] = []
+        self._items: List[dict] = []
 
         self.i = 0
         self.limit = limit
 
-    @staticmethod
-    def _next_page_request(url):
-        return Request(url, 'GET')
-
     def __aiter__(self):
         return self
 
@@ -328,19 +253,15 @@ class Paged:
 
         return item
 
-    async def _get_pages(self) -> typing.AsyncGenerator:
-        LOGGER.debug('getting first page')
-        resp = await self._do_request(self.request)
-        page = resp.json()
+    async def _get_pages(self, response) -> AsyncGenerator:
+        page = response.json()
         yield page
 
         next_url = self._next_link(page)
         while (next_url):
             LOGGER.debug('getting next page')
-            request = self._next_page_request(next_url)
-            resp = await self._do_request(request)
-            page = resp.json()
-            yield page
+            response = await self._request_fcn(url=next_url, method='GET')
+            page = response.json()
 
             # If the next URL is the same as the previous URL we will
             # get the same response and be stuck in a page cycle. This
@@ -353,6 +274,8 @@ class Paged:
                 raise PagingError(
                     "Page cycle detected at {!r}".format(next_url))
 
+            yield page
+
     def _next_link(self, page):
         try:
             next_link = page[self.LINKS_KEY][self.NEXT_KEY]
diff --git a/tests/integration/test_orders_api.py b/tests/integration/test_orders_api.py
index 00701d1..d7ddab2 100644
--- a/tests/integration/test_orders_api.py
+++ b/tests/integration/test_orders_api.py
@@ -108,8 +108,8 @@ async def test_list_orders_basic(order_descriptions, session):
         },
         "orders": [order1, order2]
     }
-    mock_resp1 = httpx.Response(HTTPStatus.OK, json=page1_response)
-    respx.get(TEST_ORDERS_URL).return_value = mock_resp1
+    mock_resp = httpx.Response(HTTPStatus.OK, json=page1_response)
+    respx.get(TEST_ORDERS_URL).return_value = mock_resp
 
     page2_response = {"_links": {"_self": next_page_url}, "orders": [order3]}
     mock_resp2 = httpx.Response(HTTPStatus.OK, json=page2_response)
@@ -188,15 +188,20 @@ async def test_list_orders_limit(order_descriptions,
 
 @respx.mock
 @pytest.mark.asyncio
-async def test_create_order(oid, order_description, order_request, session):
-    mock_resp = httpx.Response(HTTPStatus.OK, json=order_description)
-    respx.post(TEST_ORDERS_URL).return_value = mock_resp
+async def test_create_order_basic(oid,
+                                  order_description,
+                                  order_request,
+                                  session):
+    route = respx.post(TEST_ORDERS_URL)
+    route.return_value = httpx.Response(HTTPStatus.OK, json=order_description)
 
     cl = OrdersClient(session, base_url=TEST_URL)
     order = await cl.create_order(order_request)
 
     assert order == order_description
 
+    assert json.loads(route.calls.last.request.content) == order_request
+
 
 @respx.mock
 @pytest.mark.asyncio
diff --git a/tests/unit/test_http.py b/tests/unit/test_http.py
index a835e2c..5a25eb5 100644
--- a/tests/unit/test_http.py
+++ b/tests/unit/test_http.py
@@ -13,10 +13,11 @@
 # License for the specific language governing permissions and limitations under
 # the License.
 import asyncio
+import json
 import logging
 from http import HTTPStatus
 import math
-from unittest.mock import Mock, patch
+from unittest.mock import patch
 
 import httpx
 import respx
@@ -30,13 +31,6 @@ TEST_URL = 'mock://fantastic.com'
 LOGGER = logging.getLogger(__name__)
 
 
-@pytest.fixture
-def mock_request():
-    r = Mock()
-    r.http_request = httpx.Request('GET', TEST_URL)
-    yield r
-
-
 @pytest.fixture
 def mock_response():
 
@@ -198,31 +192,44 @@ async def test_session_contextmanager():
 
 @respx.mock
 @pytest.mark.asyncio
-async def test_session_request(mock_request):
+@pytest.mark.parametrize('data', (None, {'boo': 'baa'}))
+async def test_session_request_success(data):
 
     async with http.Session() as ps:
-        mock_resp = httpx.Response(HTTPStatus.OK, text='bubba')
-        respx.get(TEST_URL).return_value = mock_resp
+        resp_json = {'foo': 'bar'}
+        route = respx.get(TEST_URL)
+        route.return_value = httpx.Response(HTTPStatus.OK, json=resp_json)
+
+        resp = await ps.request(method='GET', url=TEST_URL, json=data)
+        assert resp.json() == resp_json
+
+        # the proper headers are included and they have the expected values
+        received_request = route.calls.last.request
+        assert received_request.headers['x-planet-app'] == 'python-sdk'
+        assert 'planet-client-python/' in received_request.headers[
+            'user-agent']
 
-        resp = await ps.request(mock_request)
-        assert resp.http_response.text == 'bubba'
+        if data:
+            assert received_request.headers[
+                'content-type'] == 'application/json'
+            assert json.loads(received_request.content) == data
 
 
 @respx.mock
 @pytest.mark.asyncio
-async def test_session_stream(mock_request):
+async def test_session_stream():
     async with http.Session() as ps:
         mock_resp = httpx.Response(HTTPStatus.OK, text='bubba')
         respx.get(TEST_URL).return_value = mock_resp
 
-        async with ps.stream(mock_request) as resp:
-            txt = await resp.http_response.aread()
-            assert txt == b'bubba'
+        async with ps.stream(method='GET', url=TEST_URL) as resp:
+            chunks = [c async for c in resp.aiter_bytes()]
+            assert chunks[0] == b'bubba'
 
 
 @respx.mock
 @pytest.mark.asyncio
-async def test_session_request_retry(mock_request):
+async def test_session_request_retry():
     """Test the retry in the Session.request method"""
     async with http.Session() as ps:
         route = respx.get(TEST_URL)
@@ -234,7 +241,7 @@ async def test_session_request_retry(mock_request):
         # let's not actually introduce a wait into the tests
         ps.max_retry_backoff = 0
 
-        resp = await ps.request(mock_request)
+        resp = await ps.request(method='GET', url=TEST_URL)
         assert resp
         assert route.call_count == 2
 
@@ -279,13 +286,14 @@ def test__calculate_wait():
 
 @respx.mock
 @pytest.mark.asyncio
-async def test_authsession_request(mock_request):
+async def test_authsession_request():
     sess = http.AuthSession()
-    mock_resp = httpx.Response(HTTPStatus.OK, text='bubba')
+    resp_json = {'token': 'foobar'}
+    mock_resp = httpx.Response(HTTPStatus.OK, json=resp_json)
     respx.get(TEST_URL).return_value = mock_resp
 
-    resp = sess.request(mock_request)
-    assert resp.http_response.text == 'bubba'
+    resp = sess.request(method='GET', url=TEST_URL, json={'foo': 'bar'})
+    assert resp.json() == resp_json
 
 
 def test_authsession__raise_for_status(mock_response):
diff --git a/tests/unit/test_models.py b/tests/unit/test_models.py
index c77ff2b..a95ded7 100644
--- a/tests/unit/test_models.py
+++ b/tests/unit/test_models.py
@@ -18,7 +18,6 @@ from unittest.mock import MagicMock
 import os
 from pathlib import Path
 import re
-from datetime import datetime
 
 from httpx import URL
 import pytest
@@ -29,25 +28,10 @@ from planet.exceptions import PagingError
 LOGGER = logging.getLogger(__name__)
 
 
-@pytest.fixture
-def mocked_request():
-    return models.Request('url', 'auth')
-
-
-def mock_http_response(json=None, iter_content=None, text=None):
-    m = MagicMock(name='http_response')
-    m.headers = {}
-    m.json.return_value = json or {}
-    m.aiter_content = iter_content
-    m.text = text or ''
-    return m
-
-
-def test_StreamingBody_name():
+def test_StreamingBody_name_filename():
     r = MagicMock(name='response')
-    r.request.url = URL('https://planet.com/path/to/example.tif?foo=f6f1')
-    hr = MagicMock(name='http_response')
-    hr.headers = {
+    r.url = URL('https://planet.com/path/to/example.tif?foo=f6f1')
+    r.headers = {
         'date': 'Thu, 14 Feb 2019 16:13:26 GMT',
         'last-modified': 'Wed, 22 Nov 2017 17:22:31 GMT',
         'accept-ranges': 'bytes',
@@ -55,36 +39,35 @@ def test_StreamingBody_name():
         'content-length': '57350256',
         'content-disposition': 'attachment; filename="open_california.tif"'
     }
-    r.http_response = hr
     body = models.StreamingBody(r)
     assert body.name == 'open_california.tif'
 
+
+def test_StreamingBody_name_url():
     r = MagicMock(name='response')
-    r.request.url = URL('https://planet.com/path/to/example.tif?foo=f6f1')
-    hr = MagicMock(name='http_response')
-    hr.headers = {
+    r.url = URL('https://planet.com/path/to/example.tif?foo=f6f1')
+    r.headers = {
         'date': 'Thu, 14 Feb 2019 16:13:26 GMT',
         'last-modified': 'Wed, 22 Nov 2017 17:22:31 GMT',
         'accept-ranges': 'bytes',
         'content-type': 'image/tiff',
         'content-length': '57350256',
     }
-    r.http_response = hr
     body = models.StreamingBody(r)
 
     assert body.name == 'example.tif'
 
+
+def test_StreamingBody_name_content():
     r = MagicMock(name='response')
-    r.request.url = URL('https://planet.com/path/to/noname/')
-    hr = MagicMock(name='http_response')
-    hr.headers = {
+    r.url = URL('https://planet.com/path/to/noname/')
+    r.headers = {
         'date': 'Thu, 14 Feb 2019 16:13:26 GMT',
         'last-modified': 'Wed, 22 Nov 2017 17:22:31 GMT',
         'accept-ranges': 'bytes',
         'content-type': 'image/tiff',
         'content-length': '57350256',
     }
-    r.http_response = hr
     body = models.StreamingBody(r)
 
     assert body.name.startswith('planet-')
@@ -143,7 +126,7 @@ def test__get_random_filename(content_type, check):
 
 
 @pytest.mark.asyncio
-async def test_StreamingBody_write_img(tmpdir, mocked_request, open_test_img):
+async def test_StreamingBody_write_img(tmpdir, open_test_img):
 
     async def _aiter_bytes():
         data = open_test_img.read()
@@ -154,11 +137,9 @@ async def test_StreamingBody_write_img(tmpdir, mocked_request, open_test_img):
             yield v[i * chunksize:min((i + 1) * chunksize, len(v))]
 
     r = MagicMock(name='response')
-    hr = MagicMock(name='http_response')
-    hr.aiter_bytes = _aiter_bytes
-    hr.num_bytes_downloaded = 0
-    hr.headers['Content-Length'] = 527
-    r.http_response = hr
+    r.aiter_bytes = _aiter_bytes
+    r.num_bytes_downloaded = 0
+    r.headers['Content-Length'] = 527
     body = models.StreamingBody(r)
 
     filename = Path(tmpdir) / 'test.tif'
@@ -168,111 +149,45 @@ async def test_StreamingBody_write_img(tmpdir, mocked_request, open_test_img):
     assert os.stat(filename).st_size == 527
 
 
-def test_StreamingBody_last_modified_emptyheader():
-    '''This function tests the last_modified function for an empty header, by
-    seeing if the last_modified is None.
-    '''
-    r = MagicMock(name='response')
-    r.request.url = URL('https://planet.com/path/to/example.tif?foo=f6f1')
-    hr = MagicMock(name='http_response')
-    hr.headers = {
-        'date': 'Thu, 14 Feb 2019 16:13:26 GMT',
-        'accept-ranges': 'bytes',
-        'content-type': 'image/tiff',
-        'content-length': '57350256',
-        'content-disposition': 'attachment; filename="open_california.tif"'
-    }
-    r.http_response = hr
-    body = models.StreamingBody(r)
-    output = body.last_modified()
-    expected = None
-    assert output == expected
-
-
-def test_StreamingBody_last_modified_completeheader():
-    '''This function tests the last_modified function for an existing header,
-    by comparing the last_modified date to
-    an expected output.
-    '''
-    r = MagicMock(name='response')
-    r.request.url = URL('https://planet.com/path/to/example.tif?foo=f6f1')
-    hr = MagicMock(name='http_response')
-    hr.headers = {
-        'date': 'Thu, 14 Feb 2019 16:13:26 GMT',
-        'last-modified': 'Wed, 22 Nov 2017 17:22:31 GMT',
-        'accept-ranges': 'bytes',
-        'content-type': 'image/tiff',
-        'content-length': '57350256',
-        'content-disposition': 'attachment; filename="open_california.tif"'
-    }
-    r.http_response = hr
-    body = models.StreamingBody(r)
-    output = body.last_modified()
-    expected = datetime.strptime(hr.headers['last-modified'],
-                                 '%a, %d %b %Y %H:%M:%S GMT')
-    assert output == expected
-
-
-@pytest.fixture
-def get_pages():
-    p1 = {'_links': {'next': 'blah'}, 'items': [1, 2]}
-    p2 = {'_links': {}, 'items': [3, 4]}
-    responses = [mock_http_response(json=p1), mock_http_response(json=p2)]
-
-    async def do_get(req):
-        return responses.pop(0)
-
-    return do_get
+@pytest.mark.asyncio
+async def test_Paged_iterator():
+    resp = MagicMock(name='response')
+    resp.json = lambda: {'_links': {'next': 'blah'}, 'items': [1, 2]}
 
+    async def get_response(url, method):
+        resp = MagicMock(name='response')
+        resp.json = lambda: {'_links': {}, 'items': [3, 4]}
+        return resp
 
-@pytest.mark.asyncio
-async def test_Paged_iterator(get_pages):
-    req = MagicMock()
-    paged = models.Paged(req, get_pages)
+    paged = models.Paged(resp, get_response)
     assert [1, 2, 3, 4] == [i async for i in paged]
 
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize('limit, expected', [(0, [1, 2, 3, 4]), (1, [1])])
-async def test_Paged_limit(limit, expected, get_pages):
-    req = MagicMock()
-    paged = models.Paged(req, get_pages, limit=limit)
+async def test_Paged_limit(limit, expected):
+    resp = MagicMock(name='response')
+    resp.json = lambda: {'_links': {'next': 'blah'}, 'items': [1, 2]}
+
+    async def get_response(url, method):
+        resp = MagicMock(name='response')
+        resp.json = lambda: {'_links': {}, 'items': [3, 4]}
+        return resp
+
+    paged = models.Paged(resp, get_response, limit=limit)
     assert [i async for i in paged] == expected
 
 
 @pytest.mark.asyncio
 async def test_Paged_break_page_cycle():
     """Check that we break out of a page cycle."""
+    resp = MagicMock(name='response')
+    resp.json = lambda: {'_links': {'next': 'blah'}, 'items': [1, 2]}
 
-    async def func(req):
-        return mock_http_response(json={
-            '_links': {
-                'next': 'blah'
-            }, 'items': [1, 2]
-        })
+    async def get_response(url, method):
+        return resp
 
-    req = MagicMock()
-    paged = models.Paged(req, func, limit=None)
+    paged = models.Paged(resp, get_response, limit=None)
 
     with pytest.raises(PagingError):
         [item async for item in paged]
-
-
-@pytest.mark.skip('not implemented')
-@pytest.mark.asyncio
-async def test_Paged_first_page_error():
-    """Check that we identify an error thrown by the first page upon
-    initialization."""
-
-    async def func(req):
-        return mock_http_response(404,
-                                  json={
-                                      '_links': {
-                                          'next': 'blah'
-                                      },
-                                      'items': [1, 2]
-                                  })
-
-    req = MagicMock()
-    with pytest.raises(Exception):
-        models.Paged(req, func, limit=None)
['planet/http.py', 'planet/auth.py', 'planet/clients/data.py', 'planet/clients/orders.py', 'tests/integration/test_orders_api.py', 'planet/clients/subscriptions.py', 'tests/unit/test_http.py', 'tests/unit/test_models.py', 'planet/models.py']