s3.py 4.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139
  1. """
  2. Data Transport - 1.0
  3. Steve L. Nyemba, The Phi Technology LLC
  4. This file is a wrapper around s3 bucket provided by AWS for reading and writing content
  5. TODO:
  6. - Address limitations that will properly read csv if it is stored with content type text/csv
  7. """
  8. from datetime import datetime
  9. import boto3
  10. # from boto.s3.connection import S3Connection, OrdinaryCallingFormat
  11. import numpy as np
  12. import botocore
  13. # from smart_open import smart_open
  14. import sys
  15. import json
  16. from io import StringIO
  17. import pandas as pd
  18. import json
  19. def template():
  20. return {'access_key':'access-key','secret_key':'secret-key','region':'region','bucket':'name-of-bucket','file':'file-name','chunksize':10000}
  21. class s3 :
  22. """
  23. @TODO: Implement a search function for a file given a bucket??
  24. """
  25. def __init__(self,**args) :
  26. """
  27. This function will extract a file or set of files from s3 bucket provided
  28. @param access_key
  29. @param secret_key
  30. @param path location of the file
  31. @param filter filename or filtering elements
  32. """
  33. try:
  34. self._client = boto3.client('s3',aws_access_key_id=args['access_key'],aws_secret_access_key=args['secret_key'],region_name=args['region'])
  35. self._bucket_name = args['bucket']
  36. self._file_name = args['file']
  37. self._region = args['region']
  38. except Exception as e :
  39. print (e)
  40. pass
  41. def has(self,**_args):
  42. _found = None
  43. try:
  44. if 'file' in _args and 'bucket' in _args:
  45. _found = self.meta(**_args)
  46. elif 'bucket' in _args and not 'file' in _args:
  47. _found = self._client.list_objects(Bucket=_args['bucket'])
  48. elif 'file' in _args and not 'bucket' in _args :
  49. _found = self.meta(bucket=self._bucket_name,file = _args['file'])
  50. except Exception as e:
  51. _found = None
  52. pass
  53. return type(_found) == dict
  54. def meta(self,**args):
  55. """
  56. This function will return information either about the file in a given bucket
  57. :name name of the bucket
  58. """
  59. _bucket = self._bucket_name if 'bucket' not in args else args['bucket']
  60. _file = self._file_name if 'file' not in args else args['file']
  61. _data = self._client.get_object(Bucket=_bucket,Key=_file)
  62. return _data['ResponseMetadata']
  63. def close(self):
  64. self._client.close()
  65. class Reader(s3) :
  66. """
  67. Because s3 contains buckets and files, reading becomes a tricky proposition :
  68. - list files if file is None
  69. - stream content if file is Not None
  70. @TODO: support read from all buckets, think about it
  71. """
  72. def __init__(self,**_args) :
  73. super().__init__(**_args)
  74. def _stream(self,**_args):
  75. """
  76. At this point we should stream a file from a given bucket
  77. """
  78. _object = self._client.get_object(Bucket=_args['bucket'],Key=_args['file'])
  79. _stream = None
  80. try:
  81. _stream = _object['Body'].read()
  82. except Exception as e:
  83. pass
  84. if not _stream :
  85. return None
  86. if _object['ContentType'] in ['text/csv'] :
  87. return pd.read_csv(StringIO(str(_stream).replace("\\n","\n").replace("\\r","").replace("\'","")))
  88. else:
  89. return _stream
  90. def read(self,**args) :
  91. _name = self._file_name if 'file' not in args else args['file']
  92. _bucket = args['bucket'] if 'bucket' in args else self._bucket_name
  93. return self._stream(bucket=_bucket,file=_name)
  94. class Writer(s3) :
  95. """
  96. """
  97. def __init__(self,**_args) :
  98. super().__init__(**_args)
  99. #
  100. #
  101. if not self.has(bucket=self._bucket_name) :
  102. self.make_bucket(self._bucket_name)
  103. def make_bucket(self,bucket_name):
  104. """
  105. This function will create a folder in a bucket,It is best that the bucket is organized as a namespace
  106. :name name of the folder
  107. """
  108. self._client.create_bucket(Bucket=bucket_name,CreateBucketConfiguration={'LocationConstraint': self._region})
  109. def write(self,_data,**_args):
  110. """
  111. This function will write the data to the s3 bucket, files can be either csv, or json formatted files
  112. """
  113. content = 'text/plain'
  114. if type(_data) == pd.DataFrame :
  115. _stream = _data.to_csv(index=False)
  116. content = 'text/csv'
  117. elif type(_data) == dict :
  118. _stream = json.dumps(_data)
  119. content = 'application/json'
  120. else:
  121. _stream = _data
  122. file = StringIO(_stream)
  123. bucket = self._bucket_name if 'bucket' not in _args else _args['bucket']
  124. file_name = self._file_name if 'file' not in _args else _args['file']
  125. self._client.put_object(Bucket=bucket, Key = file_name, Body=_stream,ContentType=content)
  126. pass