Based off Greg's answers to my questions I think the following will work best:

First you'll need something to wrap your open file so that it limits how much data can be read:

class FileLimiter(object):
    def __init__(self, file_obj, read_limit):
        self.read_limit = read_limit
        self.amount_seen = 0
        self.file_obj = file_obj

        # So that requests doesn't try to chunk the upload but will instead stream it:
        self.len = read_limit

    def read(self, amount=-1):
        if self.amount_seen >= self.read_limit:
            return b''
        remaining_amount = self.read_limit - self.amount_seen
        data = self.file_obj.read(min(amount, remaining_amount))
        self.amount_seen += len(data)
        return data

This should roughly work as a good wrapper object. Then you would use it like so:

 with open('my_large_file', 'rb') as file_obj:
     file_obj.seek(my_offset)
     upload = FileLimiter(file_obj, my_chunk_limit)
     r = requests.post(url, data=upload, headers={'Content-Type': 'application/octet-stream'})

The headers are obviously optional, but when streaming data to a server, it's a good idea to be a considerate user and tell the server what the type of the content is that you're sending.

Open a file and make repeatedly PUT requests to the server in chunks. To complete the upload make a POST request with the file's checksum.

Example

import hashlib
import os

import requests
from requests.auth import HTTPBasicAuth

auth = HTTPBasicAuth(username='username', password='password')

file = 'prova.txt'

size = os.path.getsize(file)

hash_md5 = hashlib.md5()

CHUNK_SIZE = 100

with open(file, 'rb') as f:
    url = 'http://localhost:8000/'
    offset = 0
    for chunk in iter(lambda: f.read(CHUNK_SIZE), b''):
        hash_md5.update(chunk)
        res = requests.put(
            url,
            data={'filename': 'my_new_file'},
            files={'file': chunk},
            headers={
                'Content-Range': f'bytes {offset}-{offset + len(chunk) -1}/{size}'
            },
            auth=auth
        )
        offset = int(res.json().get('offset'))
        url = res.json().get('url')
    finalize = requests.post(url, data={'md5': hash_md5.hexdigest()}, auth=auth)
    print(finalize.status_code)
    print(finalize.json())

For uploading a file to Google Cloud Storage in chunks, and actually benefit from using FastAPI/Starlette's request.stream()—which would allow you to receive a file/request body in chunks in your FastAPI backend; thus, avoiding loading the entire file/request body into memory (see the "Update" section of this answer for more details)—you should rather use resumable uploads. Note that, as described in this comment, one had to pay for multiple resumable upload operations in the past. However, this might have changed since then, and you should thus check for any recent updates concerning that matter.

Below is an example as given in this article, but adapted to a FastAPI application. The chunk size is set to 256 KB, as suggested in th official documentation for multiple chunk upload. Larger chunk sizes typically make uploads faster, but note that there's a tradeoff between speed and memory usage. The example is implemented using Google's offical Python Client for Google Cloud Storage (see the source code of the python-storage package along with the given samples—python-storage is also part of google-cloud-python package). For resumable uploads, Google's google-resumable-media-python package is used, which has asyncio support, but is still in development.

Since the example below does not use the asynchronous method for resumable uploads, this could cause blocking of the main thread until a chunk upload operation is completed. Thus, one might use one might have a look at this answer and the solutions provided in it, when running blocking I/O-bound or CPU-bound operations within async def endpoints (e.g., await run_in_threadpool(s.write, chunk)). However, when the async for block, in the example below, requests the next chunk from the asynchronous iterator (i.e., request.stream()), the surrounding coroutine will be suspended and pass function control back to the event loop, thus allowing other tasks in the event loop to run (regardless), until that operation is completed. Hence, even though the event loop might get blocked when the synchronous s.write(chunk) is called, the async for block would still give up time for other tasks in the event loop to run.

gcs.py

from google.auth.transport.requests import AuthorizedSession
from google.resumable_media import requests, common
from google.cloud import storage


class GCSObjectStreamUpload(object):
    def __init__(
            self, 
            client: storage.Client,
            bucket_name: str,
            blob_name: str,
            chunk_size: int=256 * 1024
        ):
        self._client = client
        self._bucket = self._client.bucket(bucket_name)
        self._blob = self._bucket.blob(blob_name)

        self._buffer = b''
        self._buffer_size = 0
        self._chunk_size = chunk_size
        self._read = 0

        self._transport = AuthorizedSession(
            credentials=self._client._credentials
        )
        self._request = None  # type: requests.ResumableUpload


    def __enter__(self):
        self.start()
        return self


    def __exit__(self, exc_type, *_):
        if exc_type is None:
            self.stop()


    def start(self):
        url = (
            f'https://storage.googleapis.com/upload/storage/v1/b/'
            f'{self._bucket.name}/o?uploadType=resumable'
        )
        self._request = requests.ResumableUpload(
            upload_url=url, chunk_size=self._chunk_size
        )
        self._request.initiate(
            transport=self._transport,
            content_type='application/octet-stream',
            stream=self,
            stream_final=False,
            metadata={'name': self._blob.name},
        )


    def stop(self):
        self._request.transmit_next_chunk(self._transport)


    def write(self, data: bytes) -> int:
        data_len = len(data)
        self._buffer_size += data_len
        self._buffer += data
        del data
        while self._buffer_size >= self._chunk_size:
            try:
                self._request.transmit_next_chunk(self._transport)
            except common.InvalidResponse:
                self._request.recover(self._transport)
        return data_len


    def read(self, chunk_size: int) -> bytes:
        to_read = min(chunk_size, self._buffer_size)
        memview = memoryview(self._buffer)
        self._buffer = memview[to_read:].tobytes()
        self._read += to_read
        self._buffer_size -= to_read
        return memview[:to_read].tobytes()


    def tell(self) -> int:
        return self._read

app.py

from fastapi import FastAPI, Request, HTTPException
from gcs import GCSObjectStreamUpload
from google.cloud import storage

app = FastAPI()

        
@app.post('/upload')
async def upload(request: Request):
    try:
        client = storage.Client()
        with GCSObjectStreamUpload(client=client, bucket_name='test-bucket', blob_name='test-blob') as s:
            async for chunk in request.stream():
                s.write(chunk)
    except Exception:
        raise HTTPException(status_code=500, detail='Something went wrong')
    
    return {"message": f"File successfuly uploaded"}

Other approaches for uploading files to Google Cloud Storage can be found in this answer.