Python.txt

GET ENVIRONMENT VARIABLE

VERSION

python --version

---------------------

conda env list
conda env create -f rait_env.yml
conda activate rait_env

-----------------

import os
os.getenv('AzureStorageConnectionString')
-------------------------

import sys
print(sys.version_info)

-------------------------

INSTALLED PACKAGES WITH VERSIONS

pip list

-------------------------------------------

GET PATH

python -c "import sys; print('\n'.join(sys.path))"

-----------------------------

MORE VERBOSITY

.vscode => launch.json => "preLaunchTask" => (Add --verbose) => "func: host start --verbose"

------------------

TRY CATCH
import sys

try:
except:
	logging.error("Error: " + str(sys.exc_info()[0]))

-----------------------------------------

DATE TIME

import datetime
x = datetime.datetime.now()

'{:02d}'.format(date.day) + "_" + '{:02d}'.format(date.month) + "_" + '{:04d}'.format(date.year)
----------------

BLOB STORAGE

Terminal => pip install azure-storage-blob

requirements.txt
	azure-functions
	azure-storage-blob

import os, uuid
from azure.storage.blob import BlobServiceClient, BlobClient, ContainerClient, __version__

blobStorageConnectionString = os.getenv('AzureStorageConnectionString')
blob_service_client = BlobServiceClient.from_connection_string(blobStorageConnectionString)

#Blob Container Instance
blobContainerName = os.getenv('ContainerName')
blob_container_client = blob_service_client.get_container_client(blobContainerName)
if blob_container_client.exists() == False:
blob_container_client.create_container()

blob = BlobClient.from_connection_string(conn_str=blobStorageConnectionString, container_name=blobContainerName, blob_name=blobName)
#if we run this function more than once a day
if blob.exists() == True:
	blob.delete_blob()

with open(fileName, "rb") as data:
	blob.upload_blob(data)
----------------------

CREATE FILE AND READ CONTENT

file = open(fileName, 'w')
file.write("Test => Workgroup => " + workGroup)
file.close()
with open(fileName, "rb") as data:
	Do something with the data
-----------------------

KEY VAULT

pip install azure-identity
pip install azure-keyvault-secrets

 KVUri = os.getenv('KeyVault_Service_URL')
        azureIdentityCredential = DefaultAzureCredential()
        KeyVaultClient = SecretClient(vault_url=KVUri, credential=azureIdentityCredential)

-----------------------

ETL

with open(path,"r") as f:
  text = f.readline().strip()  #Strips removes things like new line etc
data = [int(i) for i in text.split(",")] #creates a list of integers, instead of a list of strings
return data

---------------------

LIST COMPRENHENSION

data = [int(i) for i in text.split(",")] #creates a list of integers, instead of a list of strings
data = [i + 1 for i in data] //sums 1 to each value in the data array

------------------

CSV WRITER

def load(data, path):
  import csv
  with open(path, "w") as f:
     csv_writer = csv.writer(f)
     csv_writer.writerow(data)
  return

---------------

PREFECT 

from prefect import task, Flow

@task
def load(data, path):
   return

with Flow("nameOfTheFlow") as flow:
    

flow.visualize()
flow.run()

---------------

PREFECT PARAMETERS

from prefect import task, Flow, Parameters

inside the flow builder
path = Parameter(name="path",required=True)


flow.run(parameters={"path":"xx.csv"})

---------------------------