Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_classes(self):
test_eventer = TestEventer()
conn = RWSConnection('innovate', "FAKE_USER", "FAKE_PASS")
ODMAdapter(conn, "Mediflex", "Dev", test_eventer)
if __name__ == '__main__':
#An example
logging.basicConfig(level=logging.INFO)
from rwslib._settings import accounts
#Accounts is a dict of dicts like
# accounts = {'innovate' : {'username': 'username',
# 'password':'password'},
# 'otherurl' : {'username': 'username',
# 'password':'password'},
# }
acc = accounts['innovate']
rv = RWSConnection('innovate', acc['username'], acc['password'])
path, _ = os.path.split(os.path.realpath(__file__))
db_path = os.path.join(path, 'local_cvs.db')
#Make sqllite connection
logging.info('Opening/overwriting sqllite db %s' % db_path)
conn = sqlite3.connect(db_path)
#DBAdaptor takes care of all the database specific parts of the process
db_adaptor = SQLLiteDBAdapter(conn)
lcv = LocalCVBuilder(rv, 'SIMPLESTUDY', 'TEST', db_adaptor)
lcv.execute()
def rws(ctx, url, username, password, raw, verbose, output, virtual_dir):
if ctx.obj is None:
ctx.obj = {}
ctx.obj['URL'] = url
ctx.obj['USERNAME'] = username
ctx.obj['PASSWORD'] = password
ctx.obj['VIRTUAL_DIR'] = virtual_dir
if virtual_dir:
if username and password:
ctx.obj['RWS'] = RWSConnection(url, username, password, virtual_dir=virtual_dir)
else:
# Acceptable for UnAuth Requests
ctx.obj['RWS'] = RWSConnection(url, virtual_dir=virtual_dir)
else:
if username and password:
ctx.obj['RWS'] = RWSConnection(url, username, password)
else:
ctx.obj['RWS'] = RWSConnection(url)
ctx.obj['RAW'] = raw
ctx.obj['OUTPUT'] = output
ctx.obj['VERBOSE'] = verbose
# Variable OID required
MdsolDerivationDef("AGE", form_oid="DM", field_oid="AGE")(
# Variable OID required to be identified as a data step
MdsolDerivationStep(form_oid="DM", field_oid="DOB", data_format="StandardValue", variable_oid="DOB"),
MdsolDerivationStep(function=StepType.Age)
)
)
return odm
if __name__ == '__main__':
from rwslib import RWSConnection
from rwslib.rws_requests import PostMetadataRequest, PostDataRequest
from _settings import accounts
account = accounts['innovate']
r = RWSConnection('innovate', account['username'], account['password'])
if False: # MetaData
projectname = 'TESTSTUDY'
odm_definition = example_metadata(projectname, "Draft1")
request = PostMetadataRequest(projectname, str(odm_definition))
else: #Clinical Data
projectname = 'Mediflex'
odm_definition = example_clinical_data(projectname,"DEV")
request = PostDataRequest(str(odm_definition))
print(str(odm_definition))
response = r.send_request(request)
print(str(response))
from rwslib.rws_requests.odm_adapter import *
from rwslib.rws_requests.biostats_gateway import *
if __name__ == '__main__':
from _settings import accounts
#Accounts is a dict of dicts like
# accounts = {'innovate' : {'username': 'username',
# 'password':'password'},
# 'otherurl' : {'username': 'username',
# 'password':'password'},
# }
acc = accounts['innovate']
rave = RWSConnection('innovate', acc['username'], acc['password'])
print rave.send_request(VersionRequest(), retries=3)
print rave.send_request(BuildVersionRequest())
print rave.send_request(DiagnosticsRequest())
print rave.send_request(CacheFlushRequest()).istransactionsuccessful
print "Clinical studies request"
studies = rave.send_request(ClinicalStudiesRequest(),timeout=60)
print len(studies)
print rave.last_result.url
print rave.last_result.text
print rave.request_time
print "Metadata studies request"
m_studies = rave.send_request(MetadataStudiesRequest())
__author__ = 'andrew'
from rwslib import RWSConnection
from rwslib.rws_requests import *
from rwslib.rws_requests.odm_adapter import *
from rwslib.rws_requests.biostats_gateway import *
if __name__ == '__main__':
from _settings import accounts
acc = accounts['innovate']
rave = RWSConnection('innovate', acc['username'], acc['password'])
print(rave.send_request(VersionRequest()))
audits = rave.send_request(AuditRecordsRequest('Mediflex','Dev',startid=4000000,per_page=10000))
print rave.next_link # Get headers, next and last entries?
#print audits
while rave.next_link <> None:
rave.next(AuditRecordsRequest('Mediflex','Dev'))
print(rave.next_link)
def rws(ctx, url, username, password, raw, verbose, output, virtual_dir):
if ctx.obj is None:
ctx.obj = {}
ctx.obj['URL'] = url
ctx.obj['USERNAME'] = username
ctx.obj['PASSWORD'] = password
ctx.obj['VIRTUAL_DIR'] = virtual_dir
if virtual_dir:
if username and password:
ctx.obj['RWS'] = RWSConnection(url, username, password, virtual_dir=virtual_dir)
else:
# Acceptable for UnAuth Requests
ctx.obj['RWS'] = RWSConnection(url, virtual_dir=virtual_dir)
else:
if username and password:
ctx.obj['RWS'] = RWSConnection(url, username, password)
else:
ctx.obj['RWS'] = RWSConnection(url)
ctx.obj['RAW'] = raw
ctx.obj['OUTPUT'] = output
ctx.obj['VERBOSE'] = verbose
def rws(ctx, url, username, password, raw, verbose, output, virtual_dir):
if ctx.obj is None:
ctx.obj = {}
ctx.obj['URL'] = url
ctx.obj['USERNAME'] = username
ctx.obj['PASSWORD'] = password
ctx.obj['VIRTUAL_DIR'] = virtual_dir
if virtual_dir:
if username and password:
ctx.obj['RWS'] = RWSConnection(url, username, password, virtual_dir=virtual_dir)
else:
# Acceptable for UnAuth Requests
ctx.obj['RWS'] = RWSConnection(url, virtual_dir=virtual_dir)
else:
if username and password:
ctx.obj['RWS'] = RWSConnection(url, username, password)
else:
ctx.obj['RWS'] = RWSConnection(url)
ctx.obj['RAW'] = raw
ctx.obj['OUTPUT'] = output
ctx.obj['VERBOSE'] = verbose
ctx.obj = {}
ctx.obj['URL'] = url
ctx.obj['USERNAME'] = username
ctx.obj['PASSWORD'] = password
ctx.obj['VIRTUAL_DIR'] = virtual_dir
if virtual_dir:
if username and password:
ctx.obj['RWS'] = RWSConnection(url, username, password, virtual_dir=virtual_dir)
else:
# Acceptable for UnAuth Requests
ctx.obj['RWS'] = RWSConnection(url, virtual_dir=virtual_dir)
else:
if username and password:
ctx.obj['RWS'] = RWSConnection(url, username, password)
else:
ctx.obj['RWS'] = RWSConnection(url)
ctx.obj['RAW'] = raw
ctx.obj['OUTPUT'] = output
ctx.obj['VERBOSE'] = verbose