No project description provided
Project description
树融数据融合平台数据上传基础模块,作为是数据源和目标表的连接器,实施团队用于编写上传脚本,存储平台用于调用上传脚本。
install
download from fusionflow.upload or pip install directly:
pip install git+https://codeup.teambition.com/fusiontree/fusionplatform/fusionflow.upload.git
usage
1. write source scripts
there are two ways to write source table extract scripts
by subclass fusionflow.upload.Table:
from fusionflow.upload import Table class Employ(Table): def extract(self): yield [{"colname": "colvalue"}, ...] ...
by instantiation package-include subclass:
python from datetime import * from fusionflwo.upload.network import ApiTable def parse_api_data(data): rows = data["dataList"] {'cioStatus': 2, 'inTime': '2020-02-24 00:00:22', 'outTime': '2020-02-24 00:23:00', 'plateNo': '沪BT2741', 'recordId': '33930092'} for row in rows: d = {} for key, value in row.items(): if key in ("inTime", "outTime"): d[key] = datetime.strptime(value, "%Y-%m-%d %H:%M:%S") elif key == "recordId": d[key] = int(value) else: d[key] = value yield d parkcar_table = ApiTable(uri, table_name='parkcar_table', params={"bizDate": (lambda : (date.today() - timedelta(days=1)).strftime("%Y%m%d"))}, parser=parse_api_data)
2. parse source scripts
get table objects through scripts:
from fusionflow.upload import SrcTables src_tables = SrcTable("path/to/src/script") print(src_tables.tables)
extract data from table:
table1 = src_tables.tables[0] # get table1 field type fields = table1.fields # extract data # table object is iterator, can use in for loop for row in table1: print(row)
Django Itergration
Table class also provide a attribute django_fields to convert field type to django.db.models.fields.Field, to use this feature, use must install django moduel manually, or through:
git clone git@codeup.teambition.com:fusiontree/fusionplatform/fusionflow.upload.git pip install .[db]
the django_feilds can access from:
from fusionflow.upload.db.oracle import OraceDatabaseTable table = OracleDatabaseTable(username=user,password=passwd, host='localhost') print(table.django_fields)
3. validation
tables can add validator to check input value is business or logical leagal , validator can added through subclass parameter validators, instantiation parameter validators, or register through table.registe_validator,
must register all validators before fetch any data
currect support two types of validator:
row-leve validator (not implemented currently)
field value validator
example
subclass method:
class FileTable(Table): def extract(self): for data in emp_table(): yield data def validate_empno(self, value): assert int(value) < 7800, value
register in run time:
# at instantiation def emp_ge_7600(value): assert int(value) > 7600 f = FileTable(validate_empno=[emp_ge_7600]) # through api validator_str = """def empno_is_digit(value): assert value.isdigit() """ f.registe_validator('empno', validator_str) ```
test
use pytest test after clone this repo, run:
./test
test also itegrate with setuptools, afte git clone source run:
python setup.py test
support format
there types of easy-used subtable class provided, directly instantiate them
DatabaseTable – RDMS table source
ApiTable – application programming iterface, load from remote
FileTable – excel, csv, tsv, xml like file
contact
Project details
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distributions
Built Distribution
Hashes for fusionflow_upload-0.1.1-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 73a6c7592723957abc4b43cd20829477f4513f05a8885aca7b4ee645a505a1a3 |
|
MD5 | 0754d5610a4ecbf6d544e20e8741f8f3 |
|
BLAKE2b-256 | ef757b7503d740a32462ed4cdfc083791f3d6b9bc0451f4a999354bd9ebce0ab |