code errors

errors.py

########## Handle every case on dictionary gets and type casts.
# No:
winner_odds = schedule_json.get("sp", {}).get("main", [])
away_odds = float(winner_odds[0].get("odds", 0))
# Count the ways the above can go wrong
#  len(winner_odds) == 0
#  winner_odds[0] == {"odds": "a string"}
#  winner_odds[0] == might not be a dictionary
#  schedule_json.get("sp") might not contain a dictionary
#  schedule_json.get("sp", {}).get("main") might not contain a list
# Yes:
try:
  winner_odds = schedule_json.get("sp", {}).get("main", [])
  away_odds = float(winner_odds[0].get("odds", 0))
except (ValueError, TypeError, IndexError):


   
########## Don't hide programming errors Use asserts to force a failure ############
# No:
if oom and memory_limit:  # Disabling OOM killer should only be used with a memory limit
   result.append('--oom-kill-disable')
# Yes:
if oom:
  assert memory_limit
  result.append('--oom-kill-disable')
  

############# Always return the same type from a function
############# Naming: Is_x should always return a boolean, don’t shoehorn extra data in there.
# No:
def is_in_cache(self, param):   # Confusing name
  if cache.get('param') and datetime.utcnow() > cache.get('param'): ## If a cache timeout has happened
    cache.pop('param')
    return False            # Returns a boolean type
  return cache.get('param')  # Returns a date type

############## Try not to use Mocks in tests.
# NO:
def test_thing():
   my_thing = Mock()
   assert my_thing.marrket  # Typo in market - but assert won't catch it
# YES:
def test_thing():
   my_thing = my_thing_builder()
   assert my_thing.marrket  # assert now fails
   

############## Don't rely on default values in functions. Be explicit use Constants:
#NO:
def test_handle_duplicate_executions():
    md = get_md()
    # get_dummy_execution uses default values that we can not see from here.
    e = get_dummy_execution(20, 123456657374, 1)      # Why are these numbers so different?
    md.update_executions(1, [e])                      # What is 1 ?
    assert md.get_signed_quantities(12) == -10        # What is 12, -10 ? 

#YES:   
def test_handle_duplicate_executions():
    MARKET = 1
    CONTRACT = 12
    EXECUTION_AMOUNT = 10
    MASTER_ACCOUNT = 123456657374
    md = get_md()
    e = get_dummy_execution(
       account1=20, account2=MASTER_ACCOUNT, time=1
       market=MARKET, contract=CONTRACT, amount=EXECUTION_AMOUNT
    )
    md.update_executions(MARKET, [e])
    assert md.get_signed_quantities(CONTRACT) == -EXECUTION_AMOUNT

############## Test your functions. If you can't test them step thru several paths. Also you probably can test it really.
<no example>
   
########## Don't introduce duplicate functions. 
# Are you sure there isn't a function that already does this.
# If one is similar can it be modified to incorporate the changes you need
<no example>

########## Be careful with temporary scripts 
# Break up code in temporary scripts in to functions
# Commenting out code and intermingled asserts can make this hard to detect errors
seen = set()
for m in data['messages']:            # m & data are shit names
    m['extra_id'] = data['head_id']
    assert m['id'] not in seen        # This assert was well intentioned but it meant 
    seen.add(m['id'])                 # I didn't see the wrongly alligned save_do_db() on next line
save_to_db(m)


############ Becareful with  Transactions
try:
    with self._rw_engine.begin() as connection:
        rs = connection.execute("SQL")
        myobj.a = rs.get("a")
        myobj.b = rs.get("b")
        # myobj loaded
except Exception:
    log.exception('Unhandled exception:')  # Why log and rethrow an exception?
    raise
else:
    # This will use a different transaction to above. 
    # Are you sure this is correct? should you put in at 'myobj loaded' instead?
    myobj.load_more_from_db() 

########## Try and simplify assertions:
def func(k, n):
    assert k <= n
    assert k >= 0
    assert n >= 0
    =====
def func(k, n):    
    assert 0 <= k <= n

########## Trio / Async ordering errors
# BadCode: May deadlock at startup.
# If ObjectSendsAlert uses `alerts_send` the processes deadlocks as `process_alerts` 
# isn't running yet and so there is nothing to read off the other side of the memorychannel.
alerts_send, alerts_recv = trio.open_memory_channel[Thing](0)
async with (
   alerts_recv,
   alerts_send,
   ObjectSendsAlert(alerts_send) as sender,
):
  another_q_send, another_q_recv = trio.open_memory_channel[OtherThing](0)
  trio.start_soon(process_alerts, alerts_recv, another_q_send)