sensedeep / dynamodb-onetable / #68

Committed 18 Aug 2023 03:12AM UTC coverage: 72.741% (+0.1%) from 72.614%

Build # #68

Build Type

push

Committed by Michael O'Brien

Commit Message

Merge branch 'main' of github.com:sensedeep/dynamodb-onetable

Run Details

1121 of 1620 branches covered (69.2%)

Branch coverage included in aggregate %.

1801 of 2397 relevant lines covered (75.14%)

625.06 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.11

/src/Expression.js

/*
    Expression.js - DynamoDB API command builder

    This module converts API requests into DynamoDB commands.
*/
import {OneTableArgError, OneTableError} from './Error.js'

//  Operators used on sort keys for get/delete
const KeyOperators = ['<', '<=', '=', '>=', '>', 'begins', 'begins_with', 'between']

export class Expression {
    constructor(model, op, properties, params = {}) {
        this.init(model, op, properties, params)
        this.prepare()
    }

    init(model, op, properties, params) {
        this.model = model
        this.op = op
        this.properties = properties
        this.params = params

        this.table = model.table
        this.already = {} //  Fields already processed (index is property name).
        this.conditions = [] //  Condition expressions.
        this.filters = [] //  Filter expressions.
        this.key = {} //  Primary key attribute.
        this.keys = [] //  Key conditions.
        this.mapped = {} //  Mapped fields.
        this.names = {} //  Expression names. Keys are the indexes.
        this.namesMap = {} //  Expression names reverse map. Keys are the names.
        this.puts = {} //  Put values
        this.project = [] //  Projection expressions.
        this.values = {} //  Expression values. Keys are the value indexes.
        this.valuesMap = {} //  Expression values reverse map. Keys are the values.
        this.nindex = 0 //  Next index into names.
        this.vindex = 0 //  Next index into values.
        this.updates = {
            add: [],
            delete: [],
            remove: [],
            set: [],
        }
        this.execute = params.execute === false ? false : true
        this.tableName = model.tableName

        /*
            Find the index for this expression. Then store the attribute names for the index.
         */
        this.index = this.selectIndex(model.indexes)

        /*
            Get the request index hash/sort attributes
         */
        this.hash = this.index.hash
        this.sort = this.index.sort

        if (!this.table.client) {
            throw new OneTableArgError('Table has not yet defined a "client" instance')
        }
    }

    prepare() {
        let {op, params, properties} = this
        let fields = this.model.block.fields
        if (op == 'find') {
            this.addWhereFilters()
        } else if (op == 'delete' || op == 'put' || op == 'update' || op == 'check') {
            this.addConditions(op)
        } else if (op == 'scan') {
            this.addWhereFilters()
            /*
                Setup scan filters for properties outside the model.
                Use the property name here as there can't be a mapping.
            */
            for (let [name, value] of Object.entries(this.properties)) {
                if (fields[name] == null && value != null) {
                    this.addGenericFilter(name, value)
                    this.already[name] = true
                }
            }
        }
        this.addProperties(op, this.model.block, null, properties, this.puts)

        /*
            Emit mapped attributes that don't correspond to schema fields.
        */
        if (this.mapped) {
            for (let [att, props] of Object.entries(this.mapped)) {
                if (Object.keys(props).length != this.model.mappings[att].length) {
                    throw new OneTableArgError(
                        `Missing properties for mapped data field "${att}" in model "${this.model.name}"`
                    )
                }
            }
            for (let [k, v] of Object.entries(this.mapped)) {
                let field = {attribute: [k], name: k, filter: false}
                this.add(op, field, k, properties, v, this.puts)
            }
        }
        if (params.fields) {
            for (let name of params.fields) {
                if (op == 'batchGet') {
                    //  BatchGet params.project must provide attributes not properties
                    this.project.push(`#_${this.addName(name)}`)
                } else if (fields[name]) {
                    let att = fields[name].attribute[0]
                    this.project.push(`#_${this.addName(att)}`)
                }
            }
        }
    }

    /*
        Add properties to the command. This calls itself recursively for each schema nest level.
        At each nest level, the pathname describes the property name at that level.
        Arrays are handled linearly here. The "rec" is only used for put requests.
     */
    addProperties(op, block, pathname, properties, rec) {
        if (!properties || typeof properties != 'object') {
            return
        }
        let fields = block.fields
        for (let [name, value] of Object.entries(properties)) {
            let field = fields[name]
            let path
            if ((op == 'put')) {
                path = field ? field.attribute[0] : name
            } else if (field) {
                path = pathname ? `${pathname}.${field.attribute[0]}` : field.attribute[0]
            } else {
                path = name
            }
            if (!path) {
                this.table.log.info(`Path is null`, {op, pathname, fields, field, name, value})
            }
            if (!field) {
                if (this.model.generic) {
                    this.add(op, {attribute: [name], name}, path, properties, value, rec)
                }
                continue
            }
            let partial = this.model.getPartial(field, this.params)

            if (field.schema && value != null && partial) {
                if (field.isArray && Array.isArray(value)) {
                    rec[path] = []
                    for (let i = 0; i < value.length; i++) {
                        rec[path][i] = {}
                        let ipath = `${path}[${i}]`
                        this.addProperties(op, field.block, ipath, value[i], rec[path][i])
                    }
                } else {
                    rec[path] = {}
                    this.addProperties(op, field.block, path, value, rec[path])
                }
            } else {
                this.add(op, field, path, properties, value, rec)
            }
        }
    }

    /*
        Add a field to the command expression
     */
    add(op, field, path, properties, value, rec) {
        if (this.already[path]) {
            return
        }
        /*
            Handle mapped and packed attributes.
            The attribute[0] contains the top level attribute name. Attribute[1] contains a nested mapping name.
        */
        let attribute = field.attribute
        if (attribute.length > 1) {
            /*
                Save in mapped[] the mapped attributes which will be processed soon
             */
            let mapped = this.mapped
            let [k, v] = attribute
            mapped[k] = mapped[k] || {}
            mapped[k][v] = value
            if (op == 'put') {
                properties[k] = value
            }
            return
        }
        if (path == this.hash || path == this.sort) {
            if (op == 'find') {
                this.addKey(op, field, value)
            } else if (op == 'scan') {
                if (properties[field.name] !== undefined && field.filter !== false) {
                    this.addFilter(path, field, value)
                }
            } else if ((op == 'delete' || op == 'get' || op == 'update' || op == 'check') && field.isIndexed) {
                this.addKey(op, field, value)
            } else if (op == 'put' || (this.params.batch && op == 'update')) {
                //  Batch does not use update expressions (Ugh!)
                rec[path] = value
            }
        } else if (op == 'find' || op == 'scan') {
            //  schema.filter == false disables a field from being used in a filter
            if (properties[field.name] !== undefined && field.filter !== false) {
                if (!this.params.batch) {
                    //  Batch does not support filter expressions
                    this.addFilter(path, field, value)
                }
            }
        } else if (op == 'put' || (this.params.batch && op == 'update')) {
            //  Batch does not use update expressions (Ugh!)
            rec[path] = value
        } else if (op == 'update') {
            this.addUpdate(path, field, value)
        }
    }

    /*
        Conditions for create | delete | update
        May also be used by 'get' in fallback mode.
     */
    addConditions(op) {
        let {conditions, params} = this
        let {hash, sort} = this.index
        if (params.exists === true) {
            conditions.push(`attribute_exists(#_${this.addName(hash)})`)
            if (sort) {
                conditions.push(`attribute_exists(#_${this.addName(sort)})`)
            }
        } else if (params.exists === false) {
            conditions.push(`attribute_not_exists(#_${this.addName(hash)})`)
            if (sort) {
                conditions.push(`attribute_not_exists(#_${this.addName(sort)})`)
            }
        }
        if (params.type && sort) {
            conditions.push(`attribute_type(#_${this.addName(sort)}, ${params.type})`)
        }
        if (op == 'update') {
            this.addUpdateConditions()
        }
        if (params.where) {
            conditions.push(this.expand(params.where))
        }
    }

    /*
        Expand a where/set expression. Replace: ${var} and {value} tokens.
     */
    expand(where) {
        const expr = where
        let fields = this.model.block.fields
        //  Expand attribute references and make attribute name
        where = where.toString().replace(/\${(.*?)}/g, (match, varName) => {
            return this.makeTarget(fields, varName)
        })

        //  Expand variable substitutions
        where = where.replace(/@{(.*?)}/g, (match, value) => {
            let index
            const {substitutions} = this.params
            let name = value.replace(/^\.\.\./, '')
            if (!substitutions || substitutions[name] === undefined) {
                throw new OneTableError(`Missing substitutions for attribute value "${name}"`, {
                    expr,
                    substitutions,
                    properties: this.properties,
                })
            }
            //  Support @{...list} to support filter expressions "IN ${...list}"
            if (value != name && Array.isArray(substitutions[name])) {
                let indicies = []
                for (let item of substitutions[name]) {
                    indicies.push(this.addValue(item))
                }
                return indicies.map((i) => `:_${i}`).join(', ')
            }
            index = this.addValue(substitutions[name])
            return `:_${index}`
        })

        //  Expand value references and make attribute values. Allow new-lines in values.
        where = where.replace(/{(.*?)}/gs, (match, value) => {
            let index
            if (value.match(/^[-+]?([0-9]+(\.[0-9]*)?|\.[0-9]+)$/)) {
                index = this.addValue(+value)
            } else {
                let matched = value.match(/^"(.*)"$/)
                if (matched) {
                    index = this.addValue(matched[1])
                } else if (value instanceof Date) {
                    value = this.table.transformWriteDate(value)
                    index = this.addValue(value)
                } else if (value == 'true' || value == 'false') {
                    index = this.addValue(value == 'true' ? true : false)
                } else {
                    index = this.addValue(value)
                }
            }
            return `:_${index}`
        })
        return where
    }

    /*
        Add where filter expressions for find and scan
     */
    addWhereFilters() {
        if (this.params.where) {
            this.filters.push(this.expand(this.params.where))
        }
    }

    addFilter(pathname, field, value) {
        let {filters} = this
        /*
        let att = field.attribute[0]
        let pathname = field.pathname || att
        */
        if (pathname == this.hash || pathname == this.sort) {
            return
        }
        let [target, variable] = this.prepareKeyValue(pathname, value)
        filters.push(`${target} = ${variable}`)
    }

    /*
        Add filters when model not known
     */
    addGenericFilter(att, value) {
        this.filters.push(`#_${this.addName(att)} = :_${this.addValue(value)}`)
    }

    /*
        Add key for find, delete, get or update
     */
    addKey(op, field, value) {
        let att = field.attribute[0]
        if (op == 'find') {
            let keys = this.keys
            if (att == this.sort && typeof value == 'object' && Object.keys(value).length > 0) {
                let [action, vars] = Object.entries(value)[0]
                if (KeyOperators.indexOf(action) < 0) {
                    throw new OneTableArgError(`Invalid KeyCondition operator "${action}"`)
                }
                if (action == 'begins_with' || action == 'begins') {
                    keys.push(`begins_with(#_${this.addName(att)}, :_${this.addValue(vars)})`)
                } else if (action == 'between') {
                    keys.push(
                        `#_${this.addName(att)} BETWEEN :_${this.addValue(vars[0])} AND :_${this.addValue(vars[1])}`
                    )
                } else {
                    keys.push(`#_${this.addName(att)} ${action} :_${this.addValue(value[action])}`)
                }
            } else {
                keys.push(`#_${this.addName(att)} = :_${this.addValue(value)}`)
            }
        } else {
            this.key[att] = value
        }
    }

    /*
        Convert literal attribute names to symbolic ExpressionAttributeName indexes
     */
    prepareKey(key) {
        this.already[key] = true
        return this.makeTarget(this.model.block.fields, key)
    }

    /*
        Convert attribute values to symbolic ExpressionAttributeValue indexes
     */
    prepareKeyValue(key, value) {
        let target = this.prepareKey(key)
        let requiresExpansion = typeof value == 'string' && value.match(/\${.*?}|@{.*?}|{.*?}/)
        if (requiresExpansion) {
            return [target, this.expand(value)]
        } else {
            return [target, this.addValueExp(value)]
        }
    }

    addUpdate(pathname, field, value) {
        let {params, updates} = this
        /*
        let att = field.attribute[0]
        let pathname = field.pathname || att
        */
        if (pathname == this.hash || pathname == this.sort) {
            return
        }
        if (field.name == this.model.typeField) {
            if (!(params.exists === null || params.exists == false)) {
                //  If not creating, then don't need to update the type as it must already exist
                return
            }
        }
        if (params.remove && params.remove.indexOf(field.name) >= 0) {
            return
        }
        let target = this.prepareKey(pathname)
        let variable = this.addValueExp(value)
        updates.set.push(`${target} = ${variable}`)
    }

    addUpdateConditions() {
        let {params, updates} = this
        let fields = this.model.block.fields

        const assertIsNotPartition = (key, op) => {
            if (key == this.hash || key == this.sort) {
                throw new OneTableArgError(`Cannot ${op} hash or sort`)
            }
        }

        if (params.add) {
            for (let [key, value] of Object.entries(params.add)) {
                assertIsNotPartition(key, 'add')
                const [target, variable] = this.prepareKeyValue(key, value)
                updates.add.push(`${target} ${variable}`)
            }
        }
        if (params.delete) {
            for (let [key, value] of Object.entries(params.delete)) {
                assertIsNotPartition(key, 'delete')
                const [target, variable] = this.prepareKeyValue(key, value)
                updates.delete.push(`${target} ${variable}`)
            }
        }
        if (params.remove) {
            params.remove = [].concat(params.remove) // enforce array
            for (let key of params.remove) {
                assertIsNotPartition(key, 'remove')
                if (fields.required) {
                    throw new OneTableArgError('Cannot remove required field')
                }
                const target = this.prepareKey(key)
                updates.remove.push(`${target}`)
            }
        }
        if (params.set) {
            for (let [key, value] of Object.entries(params.set)) {
                assertIsNotPartition(key, 'set')
                const [target, variable] = this.prepareKeyValue(key, value)
                updates.set.push(`${target} = ${variable}`)
            }
        }
        if (params.push) {
            for (let [key, value] of Object.entries(params.push)) {
                assertIsNotPartition(key, 'push')
                let empty = this.addValueExp([])
                let items = this.addValueExp([].concat(value)) // enforce array on values
                const target = this.prepareKey(key)
                updates.set.push(`${target} = list_append(if_not_exists(${target}, ${empty}), ${items})`)
            }
        }
    }

    //  Translate an attribute reference to use name attributes. Works with "."
    makeTarget(fields, name) {
        let target = []
        for (let prop of name.split('.')) {
            let subscript = prop.match(/\[[^\]]+\]+/)
            if (subscript) {
                prop = prop.replace(/\[[^\]]+\]+/, '')
                subscript = subscript[0]
            } else {
                subscript = ''
            }
            let field = fields ? fields[prop] : null
            if (field) {
                target.push(`#_${this.addName(field.attribute[0])}${subscript}`)
                //  If nested schema, advance to the next level
                fields = field.schema ? field.block.fields : null
            } else {
                //  No field, so just use the property name.
                target.push(`#_${this.addName(prop)}${subscript}`)
                fields = null
            }
        }
        return target.join('.')
    }

    selectIndex(indexes) {
        let index = indexes.primary
        if (this.params.index) {
            if (this.params.index != 'primary') {
                index = indexes[this.params.index]
            }
        }
        return index
    }

    /*
        Create the Dynamo command parameters. Called from Model.run
     */
    command() {
        let {conditions, filters, key, keys, hash, model, names, op, params, project, puts, values} = this

        if (key == null && values[hash] == null && op != 'scan') {
            throw new OneTableError(`Cannot find hash key for "${op}"`, {values})
        }
        if (op == 'get' || op == 'delete' || op == 'update') {
            if (key == null) {
                return null
            }
        }
        let namesLen = Object.keys(names).length
        let valuesLen = Object.keys(values).length

        if (op == 'put') {
            puts = this.table.marshall(puts, params)
        }
        values = this.table.marshall(values, params)
        key = this.table.marshall(key, params)

        let args
        if (params.batch) {
            if (op == 'get') {
                args = {Keys: key}
            } else if (op == 'delete') {
                args = {Key: key}
            } else if (op == 'put') {
                args = {Item: puts}
            } else {
                throw new OneTableArgError(`Unsupported batch operation "${op}"`)
            }
            if (filters.length) {
                throw new OneTableArgError('Invalid filters with batch operation')
            }
        } else {
            args = {
                ConditionExpression: conditions.length ? this.and(conditions) : undefined,
                ExpressionAttributeNames: namesLen > 0 ? names : undefined,
                ExpressionAttributeValues: namesLen > 0 && valuesLen > 0 ? values : undefined,
                FilterExpression: filters.length ? this.and(filters) : undefined,
                KeyConditionExpression: keys.length ? keys.join(' and ') : undefined,
                ProjectionExpression: project.length ? project.join(', ') : undefined,
                TableName: this.tableName,
            }
            if (params.select) {
                //  Select: ALL_ATTRIBUTES | ALL_PROJECTED_ATTRIBUTES | SPECIFIC_ATTRIBUTES | COUNT
                if (project.length && params.select != 'SPECIFIC_ATTRIBUTES') {
                    throw new OneTableArgError('Select must be SPECIFIC_ATTRIBUTES with projection expressions')
                }
                args.Select = params.select
            } else if (params.count) {
                if (project.length) {
                    throw new OneTableArgError('Cannot use select and count together')
                }
                args.Select = 'COUNT'
            }
            if (params.stats || this.table.metrics) {
                args.ReturnConsumedCapacity = params.capacity || 'TOTAL' // INDEXES | TOTAL | NONE
                args.ReturnItemCollectionMetrics = 'SIZE' // SIZE | NONE
            }
            let returnValues
            if (params.return !== undefined) {
                if (params.return === true) {
                    returnValues = op === 'delete' ? 'ALL_OLD' : 'ALL_NEW'
                } else if (params.return === false || params.return == 'none') {
                    returnValues = 'NONE'
                } else if (params.return != 'get') {
                    returnValues = params.return
                }
            }
            if (op == 'put') {
                args.Item = puts
                args.ReturnValues = returnValues || 'NONE'
            } else if (op == 'update') {
                args.ReturnValues = returnValues || 'ALL_NEW'
                let updates = []
                for (let op of ['add', 'delete', 'remove', 'set']) {
                    if (this.updates[op].length) {
                        updates.push(`${op} ${this.updates[op].join(', ')}`)
                    }
                }
                args.UpdateExpression = updates.join(' ')
            } else if (op == 'delete') {
                args.ReturnValues = returnValues || 'ALL_OLD'
            }

            if (op == 'delete' || op == 'get' || op == 'update' || op == 'check') {
                args.Key = key
            }
            if (op == 'find' || op == 'get' || op == 'scan') {
                args.ConsistentRead = params.consistent ? true : false
                args.IndexName = params.index ? params.index : null
            }
            if (op == 'find' || op == 'scan') {
                args.Limit = params.limit ? params.limit : undefined
                /*
                    Scan reverse if either reverse or prev is true but not both. (XOR)
                    If both are true, then requesting the previous page of a reverse scan which is actually forwards.
                */
                args.ScanIndexForward =
                    (params.reverse == true) ^ (params.prev != null && params.next == null) ? false : true

                /*
                    Cherry pick the required properties from the next/prev param
                 */
                let cursor = params.next || params.prev
                if (cursor) {
                    let {hash, sort} = this.index
                    let start = {[hash]: cursor[hash], [sort]: cursor[sort]}
                    if (this.params.index != 'primary') {
                        let {hash, sort} = this.model.indexes.primary
                        start[hash] = cursor[hash]
                        start[sort] = cursor[sort]
                    }
                    args.ExclusiveStartKey = this.table.marshall(start, params)
                }
            }
            if (op == 'scan') {
                if (params.segments != null) {
                    args.TotalSegments = params.segments
                }
                if (params.segment != null) {
                    args.Segment = params.segment
                }
            }
        }
        //  Remove null entries
        if (args) {
            args = Object.fromEntries(Object.entries(args).filter(([, v]) => v != null))
        }

        if (typeof params.postFormat == 'function') {
            args = params.postFormat(model, args)
        }
        return args
    }

    /*
        Join the terms with 'and'
    */
    and(terms) {
        if (terms.length == 1) {
            return terms.join('')
        }
        return terms.map((t) => `(${t})`).join(' and ')
    }

    /*
        Add a name to the ExpressionAttribute names. Optimize duplicates and only store unique names once.
    */
    addName(name) {
        let index = this.namesMap[name]
        if (index == null) {
            index = this.nindex++
            this.names[`#_${index}`] = name
            this.namesMap[name] = index
        }
        return index
    }

    /*
        Add a value to the ExpressionAttribute values. Optimize duplicates and only store unique names once.
        Except for numbers because we don't want to confuse valuesMap indexes. i.e. 7 vs "7"
    */
    addValue(value) {
        let index
        if (value && typeof value != 'object' && typeof value != 'number') {
            index = this.valuesMap[value]
        }
        if (index == null) {
            index = this.vindex++
            this.values[`:_${index}`] = value
            if (value && typeof value != 'object' && typeof value != 'number') {
                this.valuesMap[value] = index
            }
        }
        return index
    }

    addValueExp(value) {
        return `:_${this.addValue(value)}`
    }
}

sensedeep / dynamodb-onetable / #68

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous