Tag: dynamodb

An AI Fantasy Football Draft Assistant
Last year I attempted to program a Fantasy Football draft assistant which took live data from ESPN’s Fantasy Platform. Boy was that a mistake…

First of all, shame on ESPN for not having an API for their Fantasy sports applications. The reverse engineered methods were not fast enough nor were they reliable. So, this year I took a new approach to building out a system for getting draft pick recommendations for my team.

I also wanted to put to use the example architecture and code I wrote the other day for the Strands SDK to work so I utilized it to build an API which would utilize the AWS Bedrock platform to analyze data and and ultimately return the best possible picks.

Here is a simple workflow of how the tool works:

I generated this with Claude AI. It is pretty OK.

The first problem I encountered was getting data. I needed two things:
1. Historical data for players
2. Projected fantasy data for the upcoming season

The historical data provides information about the players past season and the projections are for the upcoming season, obviously. The projections are useful because of any incoming rookies.

In the repository I link below I put a scripts to scrape FantasyPros for both the historical and projected data. It stores them in separate files in case you want to utilize them in a different way. There is also a script to combine them into one data source and ultimately load them into a DynamoDB table.

The most important piece of the puzzle was actually simulating the draft. I needed to create a program that would be able to track the other team’s draft picks as well as give me the suggestions and track my teams picks. This is the heart of the repository and I will be using it to get suggestions and track the draft for this coming season.

Through the application, when you issue the “next” command the application will send a request to the API with the current situation of the draft. The payload looks like this:
```
payload = {
            "team_needs": team_needs,
            "your_roster": your_roster,
            "already_drafted": all_drafted_players,
            "scoring_format": self.session.scoring_format if self.session else "ppr",
            "league_size": self.session.league_size if self.session else 12
        }
```
The “team_needs” key represents the current number of players remaining for each position. The “your_roster” position is all of the current players on my team. The other important key is “already_drafted”. This key sends all of the drafted players to the AI agent so it knows who NOT to recommend.

The application goes through all of the picks and you are able to manually enter each of the other teams picks until the draft is complete.

I’ll post an update after my draft on August 24th with the team I end up with! I still will probably lose in my league but this was fun to build. I hope to add in some sort of week-to-week management of my team as well as a trade analysis tool in the future. It would also be cool to add in some sort of analysis that could send updates to my Slack or Discord.

If you have other ideas message me on any platform you can find me on!

GitHub: https://github.com/avansledright/fantasy-football-agent
August 19, 2025
SES Monitoring
I love AWS. But one thing they don’t do is build complete tools. SES is one of them. I recently started getting emails about high usage for one of the identities that I have set up for SES. I would assume that there was a way to track usage within CloudWatch but for the life of me I couldn’t find one. So I guess that means I need to build something.

The idea here is pretty simple, within SES identities you can set up a notification. So, I created an SNS topic and subscribed all delivery notifications to the topic. Then, subscribe a Lambda function to the topic. The lambda function acts as the processor for the records then formats them in a usable way and puts them into DynamoDB. I used the identity as the primary key. The result is a simple application architecture like the below image.

Every time an email is delivered the lambda function processes the event and checks the DynamoDB table to see if we have an existing record. If the identity is already present in the table it returns the “count” value so that we can increment the value. The “destination” value appends the destination of the email being sent. Below is a sample of the code I used to put the object into the DynamoDB Table.
```
def put_dynamo_object(dynamo_object):
    count = str(dynamo_get_item(dynamo_object))
    if count == None or count == 0:
        count = str(1)
    else:
        count = int(count) + 1
    # get email address from the long string
    source_string = dynamo_object['source']
    email_match = match = re.search(r'[\w.+-]+@[\w-]+\.[\w.-]+', source_string)
    email = match.group(0)
    try:
        table.update_item(
            Key={
                    'identity': email
                    
                },
            AttributeUpdates={
                'details': {
                    'Value': {
                            'caller_identity': dynamo_object['caller_identity'],
                            'source': dynamo_object['source'],
                            'destination': dynamo_object['destination'],
                            'count': str(count)
                    }
                    
                }
            }
        )
        return True
    except ClientError as e:
        print("Failed to put record")
        print(e)
        return False
```
If you want to use this code feel free to reach out to me and I will share with you the Terraform to deploy the application and as always, reach out with questions or feedback!
November 13, 2023

Moving AWS Cloudfront Logs to DynamoDB

I think its pretty obvious that I love DynamoDB. It has become one of my favorite AWS Services and I use it almost every day at work and am getting better at using it for my personal projects as well.

I had a client approach me about getting logs from a Cloudfront Distribution. Cloudfront has a native logging function that spits out .GZ files to an S3 bucket. My client doesn’t have any sort of log ingestion service so rather than build one I decided we could parse the .GZ files and store the data into a DynamoDB table. To accomplish this I created a simple lambda:

import boto3
import gzip
import uuid
from datetime import datetime
from datetime import timedelta
import time
from botocore.exceptions import ClientError

#Creates a time to live value
def ttl_time():
    now = datetime.now()
    ttl_date = now + timedelta(90)
    final = str(time.mktime(ttl_date.timetuple()))
    return final

#Puts the log json into dynamodb:
def put_to_dynamo(record):
    client = boto3.resource('dynamodb', region_name='us-west-2')
    table = client.Table('YOUR_TABLE_NAME')
    try:
        response = table.put_item(
            Item=record
        )
        print(response)
    except ClientError as e:
        print("Failed to put record")
        print(e)
        return False

    return True
def lambda_handler(event, context):
    print(event)
    s3_key = event['Records'][0]['s3']['object']['key']
    s3 = boto3.resource("s3")
    obj = s3.Object("YOUR_BUCKET", s3_key)
    with gzip.GzipFile(fileobj=obj.get()["Body"]) as gzipfile:
        content = gzipfile.read()
    #print(content)
    my_json = content.decode('utf8').splitlines()

    my_dict = {}
    for x in my_json:
        if x.startswith("#Fields:"):
            keys = x.split(" ")
        else:
            values = x.split("\t")

    for key in keys:
        if key == "#Fields:":
            pass
        else:
            for value in values:
                my_dict[key] = value
    x = 0
    for item in keys:
        if item == "#Fields:":
            pass
        else:
            my_dict[item] = values[x]
            x +=1


    print('- ' * 20)
    myuuid = str(uuid.uuid4())
    print(myuuid)
    my_dict["uuid"] = myuuid
    my_dict['ttl'] = ttl_time()

    print(my_dict)
    if put_to_dynamo(my_dict) == True:
        print("Successfully imported item")
        return True
    else:
        print("Failed to put record")
        return False

This lambda runs every time there is an S3 object created. It takes grabs the .GZ file and parses it into a dictionary that can be imported into DynamoDB. One other thing to note is that I append a UUID so that I can help track down errors.

I wrote a simple front end for the client to grab records based on date input which writes the logs to a CSV so they can parse them on their local machines. I have a feeling we will be implementing a log aggregation server soon!

If this code helps you please share it with your friends and co-workers!

Code on Github

March 23, 2023

A Dynamo Data Migration Tool

Have you ever wanted to migrate data from one Dynamo DB table to another? I haven’t seen an AWS tool to do this so I wrote one using Python.

A quick walk through video

import sys
import boto3

## USAGE ############################################################################
## python3 dynamo.py <Source_Table> <destination table>                            ## 
## Requires two profiles to be set in your AWS Config file "source", "destination" ##
#####################################################################################
def dynamo_bulk_reader():
    session = boto3.session.Session(profile_name='source')
    dynamodb = session.resource('dynamodb', region_name="us-west-2")
    table = dynamodb.Table(sys.argv[1])

    print("Exporting items from: " + str(sys.argv[1]))

    response = table.scan()
    data = response['Items']

    while 'LastEvaluatedKey' in response:
        response = table.scan(ExclusiveStartKey=response['LastEvaluatedKey'])
        data.extend(response['Items'])

    print("Finished exporting: " + str(len(data)) + " items.")
    return data

def dynamo_bulk_writer():
    session = boto3.session.Session(profile_name='destination')
    dynamodb = session.resource('dynamodb', region_name='us-west-2')
    table = dynamodb.Table(sys.argv[2])
    print("Importing items into: " + str(sys.argv[2]))
    for table_item in dynamo_bulk_reader():
        with table.batch_writer() as batch:
            response = batch.put_item(
            Item=table_item
            )

    print("Finished importing items...")
if __name__ == '__main__':
    print("Starting Dynamo Migrater...")
    dynamo_bulk_writer()
    print("Exiting Dynamo Migrator")

The process is pretty simple. First, we get all of our data from our source table. We store this in a list. Next, we iterate over that list and write it to our destination table using the ‘Batch Writer’.

The program has been tested against tables containing over 300 items. Feel free to use it for your environments! If you do use it, please share it with your friends and link back to this article!

Github: https://github.com/avansledright/dynamo-migrate

March 31, 2022

Querying and Editing a Single Dynamo Object

I have a workflow that creates a record inside of a DynamoDB table as part of a pipeline within AWS. The record has a primary key of the Code Pipeline job. Later in the pipeline I wanted to edit that object to append the status of resources created by this pipeline.

In order to do this, I created two functions. One that first returns the item from the table and the second that actually does the update and puts the updated item back into the table. Take a look at the code below and utilize it if you need to!

import boto3 
from boto3.dynamodb.conditions import Key

def query_table(id):
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table('XXXXXXXXXXXXXX')
    response = table.query(
        KeyConditionExpression=Key('PRIMARYKEY').eq(id)
    )
    return response['Items']


def update_dynanmo_status(id, resource_name, status):
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table('XXXXXXXXXXXXX')
    items = query_table(id)
    for item in items:
        # Do your update here
        response = table.put_item(Item=item)
    return response

February 14, 2022

Tag: dynamodb

An AI Fantasy Football Draft Assistant

SES Monitoring

Moving AWS Cloudfront Logs to DynamoDB

A Dynamo Data Migration Tool

Querying and Editing a Single Dynamo Object