Файловый менеджер - Редактировать - /opt/imh-python/lib/python3.9/site-packages/libcloud/storage/drivers/__pycache__/dummy.cpython-39.pyc
Ðазад
a q�hnJ � @ s� d dl Zd dlZd dlZd dlmZ d dlmZ er@d dlmZ d dl m Z d dl m Z mZmZ d dlmZ d dlmZ d d lmZ d d lmZ G d d � d e �ZG d d� de�ZG dd� de�Zedkr�d dlZe�� dS )� N)�PY3)�b)�FileIO)� LibcloudError)�Object� Container� StorageDriver)�ContainerAlreadyExistsError)�ContainerDoesNotExistError)�ContainerIsNotEmptyError)�ObjectDoesNotExistErrorc @ s. e Zd Zd dd�Zdd� Zdd� Zd d � Zd S ) �DummyFileObject� � c C s || _ || _d S �N�� _yield_count� _chunk_len)�selfZ yield_count� chunk_len� r ��/root/rpmbuild/BUILDROOT/imh-python39-modules-3.9.7-92.el8.x86_64/opt/imh-python/lib/python3.9/site-packages/libcloud/storage/drivers/dummy.py�__init__$ s zDummyFileObject.__init__c c s* d}|| j k r&| �| j�V |d7 }qd S )Nr � )r � _get_chunkr )r �size�ir r r �read( s zDummyFileObject.readc C s dd� t �dd�D �}|S )Nc S s g | ] }t |��qS r )�str)�.0�xr r r � <listcomp>0 � z.DummyFileObject._get_chunk.<locals>.<listcomp>�a �x )�random�randint)r r �chunkr r r r / s zDummyFileObject._get_chunkc C s | j | j S r r �r r r r �__len__3 s zDummyFileObject.__len__N)r r )�__name__� __module__� __qualname__r r r r) r r r r r # s r c @ s> e Zd Zddd�Zdd� Zdd� Zdd � Zd d � Zd d � ZdS )� DummyIteratorNc C s t �� | _|pg | _d| _d S )Nr )�hashlib�md5�hash�_data� _current_item)r �datar r r r 8 s zDummyIterator.__init__c C s | j �� S r )r0 � hexdigestr( r r r � get_md5_hash= s zDummyIterator.get_md5_hashc C sB | j t| j�krt�| j| j }| j�t|�� | j d7 _ |S )Nr )r2 �lenr1 � StopIterationr0 �updater )r �valuer r r �next@ s zDummyIterator.nextc C s | � � S r )r: r( r r r �__next__I s zDummyIterator.__next__c C s d S r r r( r r r � __enter__L s zDummyIterator.__enter__c C s d S r r )r �typer9 � tracebackr r r �__exit__O s zDummyIterator.__exit__)N) r* r+ r, r r5 r: r; r< r? r r r r r- 7 s r- c @ s� e Zd ZdZdZdZdd� Zdd� Zdd � Zd'd d �Z d d� Z dd� Z dd� Z dd� Z dd� Zdd� Zd(dd �Zd)d d �Zd*d d �Zd+d!d"�Zd#d$� Zd,d%d&�Zd S )-�DummyStorageDrivera� Dummy Storage driver. >>> from libcloud.storage.drivers.dummy import DummyStorageDriver >>> driver = DummyStorageDriver('key', 'secret') >>> container = driver.create_container(container_name='test container') >>> container <Container: name=test container, provider=Dummy Storage Provider> >>> container.name 'test container' >>> container.extra['object_count'] 0 zDummy Storage Providerzhttp://example.comc C s i | _ dS )z� :param api_key: API key or username to used (required) :type api_key: ``str`` :param api_secret: Secret password to be used (required) :type api_secret: ``str`` :rtype: ``None`` N)� _containers)r Zapi_keyZ api_secretr r r r e s zDummyStorageDriver.__init__c st t � j�}t� fdd�� jD ��}d}� jD ].}� j| d }|�� D ]\}}||j7 }qFq,t|�t|�t|�d�S )aa >>> driver = DummyStorageDriver('key', 'secret') >>> driver.get_meta_data()['object_count'] 0 >>> driver.get_meta_data()['container_count'] 0 >>> driver.get_meta_data()['bytes_used'] 0 >>> container_name = 'test container 1' >>> container = driver.create_container(container_name=container_name) >>> container_name = 'test container 2' >>> container = driver.create_container(container_name=container_name) >>> obj = container.upload_object_via_stream( ... object_name='test object', iterator=DummyFileObject(5, 10), ... extra={}) >>> driver.get_meta_data()['object_count'] 1 >>> driver.get_meta_data()['container_count'] 2 >>> driver.get_meta_data()['bytes_used'] 50 :rtype: ``dict`` c s g | ]}t � j| d ��qS )�objects)r6 rA )r � containerr( r r r! � s �z4DummyStorageDriver.get_meta_data.<locals>.<listcomp>r rB )�container_count� object_count� bytes_used)r6 rA �sum�itemsr �int)r rD rE rF rC rB �_�objr r( r � get_meta_datao s � �z DummyStorageDriver.get_meta_datac c s" t | j�� �D ]}|d V qdS )a4 >>> driver = DummyStorageDriver('key', 'secret') >>> list(driver.iterate_containers()) [] >>> container_name = 'test container 1' >>> container = driver.create_container(container_name=container_name) >>> container <Container: name=test container 1, provider=Dummy Storage Provider> >>> container.name 'test container 1' >>> container_name = 'test container 2' >>> container = driver.create_container(container_name=container_name) >>> container <Container: name=test container 2, provider=Dummy Storage Provider> >>> container = driver.create_container( ... container_name='test container 2') ... #doctest: +IGNORE_EXCEPTION_DETAIL Traceback (most recent call last): ContainerAlreadyExistsError: >>> container_list=list(driver.iterate_containers()) >>> sorted([c.name for c in container_list]) ['test container 1', 'test container 2'] @inherits: :class:`StorageDriver.iterate_containers` rC N)�listrA �values�r rC r r r �iterate_containers� s z%DummyStorageDriver.iterate_containersNc C s8 | � ||�}| �|j�}| j|j d �� }| �||�S )NrB )Z_normalize_prefix_argument� get_container�namerA rN Z _filter_listed_container_objects)r rC �prefixZ ex_prefixrB r r r �iterate_container_objects� s z,DummyStorageDriver.iterate_container_objectsc C s& || j vrt| d|d��| j | d S )a� >>> driver = DummyStorageDriver('key', 'secret') >>> driver.get_container('unknown') #doctest: +IGNORE_EXCEPTION_DETAIL Traceback (most recent call last): ContainerDoesNotExistError: >>> container_name = 'test container 1' >>> container = driver.create_container(container_name=container_name) >>> container <Container: name=test container 1, provider=Dummy Storage Provider> >>> container.name 'test container 1' >>> driver.get_container('test container 1') <Container: name=test container 1, provider=Dummy Storage Provider> @inherits: :class:`StorageDriver.get_container` N��driverr9 �container_namerC )rA r )r rW r r r rQ � s �z DummyStorageDriver.get_containerc C s, |j | jvr t| d|j d��| j|j d S )a� >>> driver = DummyStorageDriver('key', 'secret') >>> driver.get_container('unknown') #doctest: +IGNORE_EXCEPTION_DETAIL Traceback (most recent call last): ContainerDoesNotExistError: >>> container_name = 'test container 1' >>> container = driver.create_container(container_name=container_name) >>> container <Container: name=test container 1, provider=Dummy Storage Provider> >>> container.name 'test container 1' >>> container.get_cdn_url() 'http://www.test.com/container/test_container_1' @inherits: :class:`StorageDriver.get_container_cdn_url` NrU �cdn_url)rR rA r rO r r r �get_container_cdn_url� s �z(DummyStorageDriver.get_container_cdn_urlc C s6 | � |� | j| d }||vr.t|d| d��|| S )a� >>> driver = DummyStorageDriver('key', 'secret') >>> driver.get_object('unknown', 'unknown') ... #doctest: +IGNORE_EXCEPTION_DETAIL Traceback (most recent call last): ContainerDoesNotExistError: >>> container_name = 'test container 1' >>> container = driver.create_container(container_name=container_name) >>> container <Container: name=test container 1, provider=Dummy Storage Provider> >>> driver.get_object( ... 'test container 1', 'unknown') #doctest: +IGNORE_EXCEPTION_DETAIL Traceback (most recent call last): ObjectDoesNotExistError: >>> obj = container.upload_object_via_stream(object_name='test object', ... iterator=DummyFileObject(5, 10), extra={}) >>> obj.name 'test object' >>> obj.size 50 @inherits: :class:`StorageDriver.get_object` rB N�� object_namer9 rV )rQ rA r )r rW r[ �container_objectsr r r � get_object� s �z DummyStorageDriver.get_objectc C s@ |j j}| j| d }|j|vr0t|jd| d��||j jd S )a >>> driver = DummyStorageDriver('key', 'secret') >>> container_name = 'test container 1' >>> container = driver.create_container(container_name=container_name) >>> container <Container: name=test container 1, provider=Dummy Storage Provider> >>> obj = container.upload_object_via_stream( ... object_name='test object 5', ... iterator=DummyFileObject(5, 10), extra={}) >>> obj.name 'test object 5' >>> obj.get_cdn_url() 'http://www.test.com/object/test_object_5' @inherits: :class:`StorageDriver.get_object_cdn_url` rB NrZ rX )rC rR rA r � meta_data)r rK rW r\ r r r �get_object_cdn_url s �z%DummyStorageDriver.get_object_cdn_urlc C sP || j v rt|d| d��ddi}t||| d�}|i d|�dd� d �| j |< |S ) aA >>> driver = DummyStorageDriver('key', 'secret') >>> container_name = 'test container 1' >>> container = driver.create_container(container_name=container_name) >>> container <Container: name=test container 1, provider=Dummy Storage Provider> >>> container = driver.create_container( ... container_name='test container 1') ... #doctest: +IGNORE_EXCEPTION_DETAIL Traceback (most recent call last): ContainerAlreadyExistsError: @inherits: :class:`StorageDriver.create_container` N�rW r9 rV rE r )rR �extrarV z http://www.test.com/container/%s� rJ )rC rB rX )rA r r �replace)r rW ra rC r r r �create_container( s � �� z#DummyStorageDriver.create_containerc C sR |j }|| jvr t|d| d��| j| }t|d �dkrFt|d| d��| j|= dS )a� >>> driver = DummyStorageDriver('key', 'secret') >>> container = Container(name = 'test container', ... extra={'object_count': 0}, driver=driver) >>> driver.delete_container(container=container) ... #doctest: +IGNORE_EXCEPTION_DETAIL Traceback (most recent call last): ContainerDoesNotExistError: >>> container = driver.create_container( ... container_name='test container 1') ... #doctest: +IGNORE_EXCEPTION_DETAIL >>> len(driver._containers) 1 >>> driver.delete_container(container=container) True >>> len(driver._containers) 0 >>> container = driver.create_container( ... container_name='test container 1') ... #doctest: +IGNORE_EXCEPTION_DETAIL >>> obj = container.upload_object_via_stream( ... object_name='test object', iterator=DummyFileObject(5, 10), ... extra={}) >>> driver.delete_container(container=container) ... #doctest: +IGNORE_EXCEPTION_DETAIL Traceback (most recent call last): ContainerIsNotEmptyError: @inherits: :class:`StorageDriver.delete_container` Nr` rB r T)rR rA r r6 r )r rC rW r r r �delete_containerH s � �z#DummyStorageDriver.delete_containerFTc C s" |t � |||d�}| jf i |��S )N)rK �response�destination_path�overwrite_existing�delete_on_failure)r Z _save_object)r rK rg rh ri Z kwargs_dictr r r �download_objectu s �z"DummyStorageDriver.download_objectc C s t � S )a >>> driver = DummyStorageDriver('key', 'secret') >>> container = driver.create_container( ... container_name='test container 1') ... #doctest: +IGNORE_EXCEPTION_DETAIL >>> obj = container.upload_object_via_stream(object_name='test object', ... iterator=DummyFileObject(5, 10), extra={}) >>> stream = container.download_object_as_stream(obj) >>> stream #doctest: +ELLIPSIS <...closed...> @inherits: :class:`StorageDriver.download_object_as_stream` )r )r rK � chunk_sizer r r �download_object_as_stream s z,DummyStorageDriver.download_object_as_streamc C s: t j�|�s td| | d��t j�|�}| j||||d�S )a+ >>> driver = DummyStorageDriver('key', 'secret') >>> container_name = 'test container 1' >>> container = driver.create_container(container_name=container_name) >>> container.upload_object(file_path='/tmp/inexistent.file', ... object_name='test') #doctest: +IGNORE_EXCEPTION_DETAIL Traceback (most recent call last): LibcloudError: >>> file_path = path = os.path.abspath(__file__) >>> file_size = os.path.getsize(file_path) >>> obj = container.upload_object(file_path=file_path, ... object_name='test') >>> obj #doctest: +ELLIPSIS <Object: name=test, size=...> >>> obj.size == file_size True @inherits: :class:`StorageDriver.upload_object` zFile %s does not exist)r9 rV �rC r[ r ra )�os�path�existsr �getsize� _add_object)r � file_pathrC r[ ra Z verify_hash�headersr r r r � upload_object� s � �z DummyStorageDriver.upload_objectc C s t |�}| j||||d�S )a >>> driver = DummyStorageDriver('key', 'secret') >>> container = driver.create_container( ... container_name='test container 1') ... #doctest: +IGNORE_EXCEPTION_DETAIL >>> obj = container.upload_object_via_stream( ... object_name='test object', iterator=DummyFileObject(5, 10), ... extra={}) >>> obj #doctest: +ELLIPSIS <Object: name=test object, size=50, ...> @inherits: :class:`StorageDriver.upload_object_via_stream` rm )r6 rr )r �iteratorrC r[ ra rt r r r r �upload_object_via_stream� s �z+DummyStorageDriver.upload_object_via_streamc C s0 |j j}|j}| j||d�}| j| d |= dS )a` >>> driver = DummyStorageDriver('key', 'secret') >>> container = driver.create_container( ... container_name='test container 1') ... #doctest: +IGNORE_EXCEPTION_DETAIL >>> obj = container.upload_object_via_stream(object_name='test object', ... iterator=DummyFileObject(5, 10), extra={}) >>> obj #doctest: +ELLIPSIS <Object: name=test object, size=50, ...> >>> container.delete_object(obj=obj) True >>> obj = Object(name='test object 2', ... size=1000, hash=None, extra=None, ... meta_data=None, container=container,driver=None) >>> container.delete_object(obj=obj) #doctest: +IGNORE_EXCEPTION_DETAIL Traceback (most recent call last): ObjectDoesNotExistError: @inherits: :class:`StorageDriver.delete_object` )rW r[ rB T)rC rR r] rA )r rK rW r[ r r r � delete_object� s �z DummyStorageDriver.delete_objectc C sh | � |j�}|pi }|�di �}|�dd|�dd� i� t|||d ||| d�}|| j|j d |< |S )Nr^ rX z http://www.test.com/object/%srb rJ )rR r ra r0 r^ rC rV rB )rQ rR �getr8 rc r rA )r rC r[ r ra r^ rK r r r rr � s � �z DummyStorageDriver._add_object)NN)FT)N)NTN)NN)N)r* r+ r, �__doc__rR Zwebsiter rL rP rT rQ rY r] r_ rd re rj rl ru rw rx rr r r r r r@ S s4 ( � ! - � � � r@ �__main__) �os.pathrn r% r. Zlibcloud.utils.py3r r �ior �fileZlibcloud.common.typesr Zlibcloud.storage.baser r r Zlibcloud.storage.typesr r r r r �objectr- r@ r* �doctest�testmodr r r r �<module> s*
| ver. 1.1 | |
.
| PHP 8.3.30 | Ð“ÐµÐ½ÐµÑ€Ð°Ñ†Ð¸Ñ Ñтраницы: 0 |
proxy
|
phpinfo
|
ÐаÑтройка