Select to view content in your preferred language

Querying Date Created Field Returns Value Even If Record Count is Zero (0)

101
1
Monday
McKinneyPatrick_PA
Occasional Contributor

I have a Python script that fetches various attributes from surveys and their associated feature service for both ArcGIS Online and ArcGIS Enterprise.  I am encountering a behavior where when querying the feature service, a value from the date created field ("created_date" in Portal; "CreationDate" in ArcGIS Online) is returned even if the record count is zero.

I'm not sure how there can be a date for records last added even if there is no record count.  Furthermore, this date does not match the date for then the survey was created, or when it was last modified.  I have provided code as an attachment.  

1 Reply
HaydenWelch
MVP Regular Contributor

Something I tend to do with scripts like this is make it digestable. Here's a rough re-write of the get_details script that breaks out the process into functions:

from typing import TypedDict, TYPE_CHECKING
from arcgis.gis import GIS, ContentManager, Item, Group, Layer

if TYPE_CHECKING:
    from arcgis.gis._impl._content_manager import SharingManager, SharingGroupManager

import sys
import logging
from functions import get_local_date, set_groups_data, get_edit_dates, extract_user_props

SurveyRecord = TypedDict(
    'SurveyRecord',
    {
        "Survey_ID": str, #survey_id,
        "Title": str, #survey_title,
        "Owner": str, #survey_owner,               
        "Survey_Created": str, #get_local_date(survey.created),                
        "Survey_Last_Modified_By_Owner": str, #get_local_date(survey.modified),  
        "Summary": str, #survey_summary,
        "Description": str, #survey_desc,                              
        "Sharing_Access": str, #survey_access,                   
        "Record_Count": int, #record_count,
        "Records_Last_Added": str, #records_last_added,
        "Survey_Table_Schema_Last_Modified_By_Owner": str, #field_schema_last_modified,
        "Shared_with_Groups": str, #set_groups_data(group_names), 
        "Group_Members": str, #set_groups_data(formatted_group_members) if formatted_group_members else set_groups_data(group_members),
        "Count_Group_Members": int,#group_members_count,
        "Platform": str, #platform
    },
    total=False # Allow an empty record
)

def get_survey_props(gis: GIS, platform: str) -> list[SurveyRecord] | None:
    c_manager: ContentManager = gis.content
    surveys: list[Item] = c_manager.search(query='type:"Form"', max_items=5000)
    
    # Handle no surveys
    if not surveys:
        logging.info(f'No surveys found on {platform}')
        return None
    
    logging.info(f'generated list of surveys for {platform}')
    
    date_field = 'created_date' if platform == 'Portal' else 'CreationDate'
    
    # Get a list of survey data in the SurveyRecord format or None
    recs = [get_survey_data(gis, survey, date_field, platform) for survey in surveys]
    
    # Log no records found
    if not recs:
        logging.info(
            f'No survey records found!'
        )
        return None
    
    # Filter out survey records that couldn't be processed
    return list(filter(None, recs))

def get_survey_data(gis: GIS, survey: Item, date_field: str, platform: str) -> SurveyRecord | None:
    # Initialize a record
    survey_record = SurveyRecord()
    
    # Get base attributes
    survey_record['Survey_ID'] = get_survey_attr(survey, 'id', 'UNK')
    survey_record['Title'] = get_survey_attr(survey, 'title', 'UNK')
    survey_record['Owner'] = get_survey_attr(survey, 'owner', 'UNK')
    survey_record['Summary'] = get_survey_attr(survey, 'snippet', 'UNK')
    survey_record['Description'] = get_survey_attr(survey, 'description', 'UNK')
    survey_record['Sharing_Access'] = get_survey_attr(survey, 'access', 'UNK')
    survey_record['Platform'] = platform
    
    # Find Groups
    s_manager: SharingManager = survey.sharing
    g_manager: SharingGroupManager= s_manager.groups
    groups: list[Group] = g_manager.list()
    if len(groups) == 0:
        logging.info(f'Failed to find groups for {survey.title}')
        return None # Return partial rec?
    
    group_names, group_members, group_member_count = process_groups(groups)
    survey_record['Shared_with_Groups'] = set_groups_data(group_names)
    survey_record['Group_Members'] = set_groups_data(group_members)
    survey_record['Count_Group_Members'] = group_member_count
    
    # Process service related info
    service = (
        get_service_by_related(get_related_services(survey), survey) 
        or get_service_by_id(gis, survey)
    )
    
    if not service.layers or len(service.layers) == 0:
        logging.info(
            f'Invalid layer configuration for Survey {survey.title}'
        )
        return None # Return partial rec?
    
    # Calculated by process_service
    #survey_record['Record_Count'] = None
    #survey_record['Records_Last_Added'] = None
    #survey_record['Survey_Table_Schema_Last_Modified_By_Owner'] = None
    return process_service(service, date_field, survey_record)

def get_survey_attr(survey: Item, attr: str, default: str='UNK') -> str:
    val: str|None = getattr(survey, attr, None)
    return val.lstrip().rstrip() if val else default

def get_related_services(survey: Item) -> list[Item] | None:
    return [
        rel for rel in survey.related_items('Survey2Service', 'forward') 
        if rel.type == 'Feature Service'
    ] or None

def get_service_by_related(related_services: list[Item]|None, survey: Item) -> Item | None:
    if not related_services:
        logging.info(
            f'No related services found for {survey.title} falling back to serviceItemID...'
        )
        return None
    
    service = None
    for rel_service in related_services:
        if rel_service.type != 'Feature Service':
            continue
        service = rel_service
    return service

def get_service_by_id(gis: GIS, survey: Item) -> Item | None:
    service_id: str = survey.properties.get("serviceItemId")
    
    if not service_id:
        logging.info(
            f'Failed to get Service Item ID for {survey.title}'
        )
        return None
    
    c_manager: ContentManager = gis.content
    service = c_manager.get(service_id)
    
    if service.type != 'Feature Service':
        logging.info(
            f'Failed to find a valid Feature Service for {survey.title}'
        )
    
    return service

def process_groups(groups: list[Group]) -> tuple[str, str, int]:
    if not groups or len(groups) == 0:
        return 'None', 'None', 0
    
    group_names = [group.title for group in groups]
    
    group_members = []
    for group in groups:
        # get list of members
        # {'owner': 'user', 'admins': ['user'], 'users': ['user']}
        members = group.get_members()
        # add owner
        group_members.append(members['owner'])
        # add admins
        for member in members['admins']:
            group_members.append(member)
        # add users
        for member in members['users']:
            group_members.append(member)
    # remove duplicates
    group_members = sorted(set(group_members))
    group_members_count += len(group_members)
    formatted_group_members = extract_user_props(group_members)
    return group_names, formatted_group_members or group_members, group_members_count
    
def process_service(service: Item, date_field: str, survey_record: SurveyRecord) -> SurveyRecord:    
    layer: Layer = service.layers[0]
    record_count: int = layer.query(return_count_only=True)
    dates = get_edit_dates(layer.properties)                              

    # query most recent date that a record was added                     
    query_result = layer.query(
        where="1=1", # Filter only 'new' records here?
        out_fields=date_field,
        order_by_fields=f"{date_field} DESC",
        return_geometry=False,
        result_record_count=1
    )
    
    if not query_result.features:
        return survey_record # Return partial? or None?
    
    recent_date = query_result.features[0].attributes[f'{date_field}']
    records_last_added = get_local_date(recent_date)
    #records_last_added = get_local_date(dates['dataLastEditDate'])
    field_schema_last_modified = get_local_date(dates['schemaLastEditDate'])
    
    survey_record['Record_Count'] = record_count
    survey_record['Records_Last_Added'] = records_last_added
    survey_record['Survey_Table_Schema_Last_Modified_By_Owner'] = field_schema_last_modified
    survey_record['Shared_with_Groups'] = None
    survey_record['Group_Members'] = None
    survey_record['Count_Group_Members'] = None
    
    return survey_record

 

This way you can more easily narrow down exactly which logical part of your script is causing issues. I trued to leave some comments in places that could be causing issues.

0 Kudos