Different from the coding convention as that deals with style. This deals with 'how todos'.
- Classes
- Context Managers
- Decorators
- Dictionaries
- Enums
- Error Handling
- File System
- Functions
- Iterators
- json
- Loops
- Linting
- Mocking
- OS
- Project Structure
- Operators
- Sets
- Stdin
- Strings
- String Formatting
- Testing
- Threading
- Twisted
- Types
- virtualenv
Useful for creating an instance of class with different parameters.
@staticmethod- declares the method as static, you can call the function without instantiating the object. (Class.from_client_order(<some_client_order>))
@staticmethod
def from_client_order(client_order):
"""
Static method to construct the controller using a request.
"""
controller = ClientOrderStatusController()
try:
controller.parse_client_order(client_order)
except ClientOrderStatusControllerError as e:
raise e
Context Managers allows you to release and acquire resources more succintly.
This example can reache a point where a mutex lock is never release if an exception is raised.
from threading import Lock
lock = Lock()
def do_something_dangerous():
lock.acquire()
raise Exception('oops I forgot this code could raise exceptions')
lock.release()
try:
do_something_dangerous()
except:
print('Got an exception')
lock.acquire()
print('Got here')
With a context manager, using the with pattern, the lock is released when programs moves out of the indentation.
from threading import Lock
lock = Lock()
def do_something_dangerous():
with lock:
raise Exception('oops I forgot this code could raise exceptions')
try:
do_something_dangerous()
except:
print('Got an exception')
lock.acquire()
print('Got here')
from contextlib import contextmanager
# Using @contextmanager
# - file_handler() is exectued as __enter()__
# - yield returns the file_handler when `with open_file()` is called.
# - file_handler.close() is called when the program moves out of the indentation block.
@contextmanager
def open_file(path, mode):
file_handler = open(path, mode)
yield file_handler
file_handler.close()
files = []
for x in range(10000):
print('range called {}'.format(x))
with open_file('foo.txt', 'w') as infile:
files.append(infile)
for f in files:
if not f.closed:
print('not closed')
# Using @contextmanager
# - Prints the name as __enter()__
# - yields nothing meaning returns nothing on `with tag()` and allows the indented
# block of code to execute.
# - on __exit()__, prints another tag.
@contextmanager
def tag(name):
print("<%s>" % name)
yield
print("</%s>" % name)
with tag("h1"):
print('foo')
@contextmanager
def open_something(name):
print('this was __enter__ and about to yield bar')
yield 'bar'
print('this was __exit__')
with open_something('foo') as bar:
print('this was yielded: {}'.format(bar))
print('this is code beining excuted on yield')
Used to add functionality to functions without hard coding.
1. Executes the Check function
2. Returns the inside function
3. Inside function calls func if b is not 0
def check(func):
def inside(a, b):
if b == 0:
print("Can't divide by 0")
return
func(a, b)
return inside
@check
def div(a, b):
return a / b
print(div(10, 0))
valid = {'buy':0, 'sell':0}
valid = dict(buy=0, sell=0)
valid = {'buy':0, 'sell':0}
keys = list(valid.keys())
> ['buy', 'sell]
valid = dict(buy=0, sell=0)
for key, value in valid.items():
print(key, value)
Each enum uses the key as it's variant name.
The numbers that increment sequantially have no significance to the value of the enum.
import enum
class SomeEnum(enum.Enum):
foo = 1
bar = 2
Accesses the string representation of the enum variant.
SomeEnum.foo.name
Enums can be iterated
for e in enum:
print(e.name)
Check if a value is a variant of an enum
if value.lower() not in SomeEnum.__members__:
When opening files, if applicable use the with open('some-file', 'some-access') as f:.
The reason being, the file handler will only exist in the life of the indentation. After the program moves out of the indented code, the file handler will be automatically closed.
Example of Python calls close(), on file handler using context managers:
__exit__() is called when the object exists from memory.
class File():
def __init__(self, filename, mode):
self.filename = filename
self.mode = mode
def __enter__(self):
self.open_file = open(self.filename, self.mode)
return self.open_file
def __exit__(self, *args):
self.open_file.close()
files = []
for _ in range(10000):
with File('foo.txt', 'w') as infile:
infile.write('foo')
files.append(infile)
w+ - Create file if it doesn't exist and write to file.
file_handler = open(
'{}/some-file.json'.format(folder_name), 'w+'
)
r - Read only.
file_handler = open(
'{}/some-file.json'.format(folder_name), 'r'
)
a - Read only.
file_handler = open(
'{}/some-file.json'.format(folder_name), 'a'
)
w - Write only.
file_handler = open(
'{}/some-file.json'.format(folder_name), 'w'
)
Create a custom error for each class or file scope.
- Inherent from the base error
- Create custom errors from the custom base error
- When checking for errors, only check the custom base error
class MyException(BaseException):
pass
class NonPositiveIntegerError(MyException):
pass
class TooBigIntegerError(MyException):
pass
...
try:
average(1, -1)
except MyException as e:
print(e)
try:
result = format_msg(cli_impl.get_enode_address())
except Exception as e:
raise Exception(e)
return result
TypeError: Wrong type passed.
ValueError: Wrong value passed.
some_func(something, *keys):
for k in keys:
...
Returns a list given an iterable.
x = dict(buy=0, sell=0)
items = list(x.keys())
print(items)
> ['buy', 'sell']
NOTE:
- The function should not be called when passing into map.
- To retain changes, wrap the map in list.
list(map(some-function, some-list))
Standard library for serializing and deserializing json.
Serializes an object to a json formatted string.
import json
req = {'hello': 'world'}
req_json = json.dumps(req)
x = dict(buy=0, sell=0)
items = [key for key, _ in x.items()]
print(items)
> ['buy', 'sell']
Currently using Black
$ pip install black
$ black <some-file>.py
from unittest import mock
m = mock.Mock()
m.foo = 'bar'
assert m.foo == 'bar'
m.configure_mock(bar='baz')
assert m.bar == 'baz'
Can set returned values for a mock function call and side effects as an iterable.
from unittest import mock
‼ m = mock.Mock()
# Assigns a return value of the mock function.
# When the mock function is called, it will return 42.
m.return_value = 42
assert m() == 42
# Creates an iterable using side_effect.
m.side_effect = ['foo', 'bar', 'baz']
assert m() == 'foo'
assert m() == 'bar'
assert m() == 'baz'
✖ with pytest.raises(StopIteration):
m()
import os
os.remove('some-file')
Removes all files recursively but DOES NOT delete the parent folder.
files = glob.glob('some-folder/*')
list(map(os.remove, files))
Cleaner way to use and like statements.
Before:
if current_time > start_time and current_time < end_time:
After:
if start_time < current_time < end_time:
return False if error == {} else True
With python packages, the import module scope is limited to the root level.
We can use sys.path.append("..") to enable an import from an above level.
import sys
sys.path.append('..')
Allows the next import to be able to import from the path above.
This can also be used on more than 1 level.
sys.path.append('..')
sys.path.append('..')
user_input = input('Enter input: ')
Similar to Dictionaries but contains one item instead of a k,v pair, the same concept as a mathematical set.
Use the not expression to check for an empty string.
if not some_str:
hello = " hello world\n\n\n\n\n"
print(hello.strip())
> "hello world"
some_str.upper()
some_str.lower()
a_str = 'hello '\
'world, '\
'how '\
'are '\
'you?'
Format using {}.format(...).
'{}/enode-address.txt'.format(self.hello)
'{} {}!'.format(hello, name)
Repeats the 0 index variable self.name twice because we declare the 0 index twice.
'docker build -t {0}:latest {0}'.format(self.container_name)
Replaces a given character with another.
x = "Hello-World"
x.replace("-", "")
A recipe for testing decorators.
Using a mock/empty function that applies the decorator.
def a_decorator(func):
def inside(a):
<do stuff>
return func(a)
return inside
TESTS:
import a_decorator
@a_decorator
def mock_function():
pass
def test_decorator():
with pytest.raises(Exception):
mock_function()
A very useful package for testing time related functions. We can set datetime.now() to a specific date using freezegun.
from freezegun import freeze_time
@freeze_time("2019-12-25")
def test_something():
datetime.now() is "2019-12-25"
Currently using pytest to run tests.
Setup:
$ pip install pytest
Run testes:
$ pytest
with pytest.raises(IOError):
some_func_that_should_raise_IOError()
@contextmanager
def not_raises():
try:
yield
except Exception as e:
raise pytest.fail(f'exception was raised: {e}')
Check for Equality (a == b)
assertEqual(a, b)
Check for Inequality (a != b)
assertNotEqual(a, b)
Check for Bool
assertTrue(x)
assertFalse(x)
Check a is b (a is b)
assertIs(a, b)
Check for is not (a is not b)
assertIsNot(a, b)
Check for None
assertIsNone(x)
Check for Not None
assertIsNotNone(x)
Check a is in b
assertIn(a, b)
Check a is not in b
assertNotIn(a, b)
Check a is an instance of b
assertIsInstance(a, b)
Check a is not an instance of b
assertNotIsInstance(a, b)
Check a exception is raised
assertRaises(exc, fun, args, *kwds)
Assert Greater/Equal, Lesser/Equal
assertGreater(a, b)
assertGreaterEqual(a, b)
assertLess(a, b)
assertLessEqual(a, b)
Check dictionary contains
assertDictContainsSubset(a, b)
x = threading.Thread(target=some-func, args(some-arg, ))- Creates a block thread that executes a function.x.start()- Starts the thread.
import logging
import threading
import time
# This thread blocks until complete.
def thread_function(name):
✖ logging.info(f'Thread {name}: starting')
time.sleep(2)
logging.info(f'Thread {name}: finishing')
if __name__ == '__main__':
format = '%(asctime)s: %(message)s'
logging.basicConfig(format=format, level=logging.INFO, datefmt='%H:%M:%S')
logging.info('Main : before creating thread')
x = threading.Thread(target=thread_function, args=(1, ))
logging.info('Main : before running thread')
x.start()
logging.info('Main : wait for the thread to finish')
logging.info('Main : all done')
Blocks the main thread until the thread has finished.
Stops the programming from exiting.
x = threading.Thread(target=thread_function, args=(1,))
print('do stuff and wait')
x.join()
print('carrying on with the rest of the main thread')
A non blocking daemon thread, runs in the background.
threading.Thread(target=some-func, args=(1,), daemon=True)
Runs a thread given an interval.
from threading import Timer
def thread_example():
print('thread exampled called!')
Timer(2.0, thread_example).start()
print('starting thread_example')
thread_example()
# Runs immediately after on the main thread.
print('this is run after thread_example')
A very useful pattern for waiting for an event to occur
def logon_connection_test():
"""
Part of the Session Level Tests.
Tests that we can connect to the exchange and receive a `Logon` and a
`Logout`.
We need x3 Logon Events to occur:
- FIX.4.2:307u->JLQD
- FIX.4.2:306u->JLQD
- FIX.4.2:305u->JLQD
And we need to be able to receive a Logout message to gracefully close the
connection.
"""
test_complete_event = threading.Event()
class MockFixModel(FixModel):
logon_events = []
logout_event = None
def onLogon(self, sessionID):
print(f"LogOn called: {sessionID}")
self.logon_events.append(sessionID)
if len(self.logon_events) >= 3:
test_complete_event.set()
return
def onLogout(self, sessionID):
self.logout_event = sessionID
return
print("-------- FIX LOGON TEST STARTING -------")
fix_model = MockFixModel()
fix_client = i2FixConnMan(fix_model)
thread = threading.Thread(target=fix_client.connect)
thread.start()
test_complete_event.wait()
fix_client.logout()
assert fix_model.logon_queue is not None
assert len(fix_model.logon_queue) == 3
assert fix_model.logout_event is not None
A library that is used in the Autobahn library.
It's a websocket library.
Twisted runs asynchronously using an event loop model.
Writing and Reading from a DB needs to be asynchronous due to the event loop. Making a synchronous call will block the websocket server.
This is achieved through the use of returned deffers which are basically generator callbacks.
Twisted provides an api adbapi to create a connection pool that creates a connection to the db in a separate thread
to not block the event loop.
Below is an example of using twisted to make db calls:
from twisted.enterprise import adbapi
db_pool = adbapi.ConnectionPool(
"psycopg2cffi",
host=os.environ["POSTGRES_HOST"],
port=5432,
database=os.environ["POSTGRES_DB"],
user=os.environ["POSTGRES_USER"],
password=os.environ["POSTGRES_PASSWORD"],
)
def async_get_admins(self):
"""
Async method that wraps getting all the admins from the DB and then
executes the callback function to update a list of admins.
"""
admins = []
callback = lambda x: [admins.append(a[0]) for a in x]
self.get_ws_admins().addCallback(callback)
return admins
def get_ws_admins(self):
"""
Retreives a list of UUIDs corresponding to the Admins that have permissions
to view websocket messages.
"""
q = """
SELECT *
FROM party_permissions
WHERE perm_name = 'admin-ws-recv-all'
"""
return db_pool.runQuery(q)
if isinstance(some_var, str):
if not isinstance(some_var, str):
https://springmerchant.com/bigcommerce/psycopg2-virtualenv-install-pg_config-executable-not-found/