mirror of
https://github.com/datahub-project/datahub.git
synced 2026-01-05 06:16:37 +00:00
28 lines
688 B
Python
28 lines
688 B
Python
import requests
|
|
from tests.consistency_utils import wait_for_writes_to_sync
|
|
|
|
|
|
class CustomSession(requests.Session):
|
|
"""
|
|
Create a custom session to add consistency delay on writes
|
|
"""
|
|
|
|
def post(self, *args, **kwargs):
|
|
response = super(CustomSession, self).post(*args, **kwargs)
|
|
if "/logIn" not in args[0]:
|
|
print("sleeping.")
|
|
wait_for_writes_to_sync()
|
|
return response
|
|
|
|
|
|
def post(*args, **kwargs):
|
|
response = requests.post(*args, **kwargs)
|
|
if "/logIn" not in args[0]:
|
|
print("sleeping.")
|
|
wait_for_writes_to_sync()
|
|
return response
|
|
|
|
|
|
def get(*args, **kwargs):
|
|
return requests.get(*args, **kwargs)
|