DEV Community 👩‍💻👨‍💻

Yao-Ching Huang
Yao-Ching Huang

Posted on • Updated on

Python :: 深入理解moto實作細節

實作分析基於moto版本3.1.x。

Amazon S3(後續簡稱S3)是目前相當知名的雲端物件儲存服務之一。我們資料科學團隊的專案也經常整合S3服務應用於訓練資料以及模型存放。但涉及程式碼測試,和S3的互動部分該如何進行測試?

spulec/moto套件提供了本地模擬Amazon服務的伺服器端的行為。我們在Pytest內只需簡單設定fixture即可。

這篇文章紀錄moto原始碼如何模擬S3特性,包含上傳、下載、版本紀錄等特性。

Amazon S3

Bucket

FakeBucket表示每個新增的bucket。

https://github.com/spulec/moto/blob/3.1.6/moto/s3/models.py#L847

如何實作create_bucket

每個bucket都會對應一個FakeBucket instance。

並且用buckets(dict object)紀錄每個的bucket對應的instance。

def create_bucket(self, bucket_name, region_name):
    ...
    new_bucket = FakeBucket(name=bucket_name, region_name=region_name)

    self.buckets[bucket_name] = new_bucket
    return new_bucket
Enter fullscreen mode Exit fullscreen mode

如何實作S3物件的多版本特性

_VersionedKeyStore 實現dict一個key可以儲存多個value,但是get method只會回傳最新的value。

https://github.com/spulec/moto/blob/3.1.6/moto/s3/utils.py#L102

Object

FakeKey表示每個新增的object。

https://github.com/spulec/moto/blob/3.1.6/moto/s3/models.py#L102

如何實作put_object

https://github.com/spulec/moto/blob/3.1.6/moto/s3/models.py#L1608

上傳的物件對應至參數value

如何實作delete_object

dictpop模擬刪除物件的行為。

https://github.com/spulec/moto/blob/3.1.6/moto/s3/models.py#L1989

moto如何避免吃光記憶體 - SpooledTemporaryFile

由SpooledTemporaryFile instance負責處理Python object的儲存方式。當物件大小超過max_size就會寫入硬碟。

https://docs.python.org/3/library/tempfile.html#tempfile.SpooledTemporaryFile

This class operates exactly as TemporaryFile() does, except that data is spooled in memory until the file size exceeds max_size, or until the file’s fileno() method is called, at which point the contents are written to disk and operation proceeds as with TemporaryFile().


self._value_buffer = tempfile.SpooledTemporaryFile(self._max_buffer_size)

...

@property
def value(self):
    self.lock.acquire()
    self._value_buffer.seek(0)
    r = self._value_buffer.read()
    r = copy.copy(r)
    self.lock.release()
    return r

...

@value.setter
def value(self, new_value):
    self._value_buffer.seek(0)
    self._value_buffer.truncate()

    # Hack for working around moto's own unit tests; this probably won't
    # actually get hit in normal use.
    if isinstance(new_value, str):
        new_value = new_value.encode(DEFAULT_TEXT_ENCODING)
    self._value_buffer.write(new_value)
    self.contentsize = len(new_value)
Enter fullscreen mode Exit fullscreen mode

Top comments (0)

⬇️ The only reason people scroll to the bottom...

...is because they want to read more. Create an account to bookmark, comment, and react to articles that interest you.