https://prefect.io logo
Join Slack
Powered by
Hi everyone, I just published a blogpost about a ...
# show-and-tell
f
Hi everyone, I just published a blogpost about a personal project 🤓 where I used Prefect to build a fully automated Medium stats pipeline 📊 to track and improve my writing performance on Medium. A big shoutout to @Dylan and @Jimmy Le for helping me with the Selenium issue and also to @Chris White for resetting my Cloud account after a stupid mistake! I hope you enjoy it: https://felix-vemmer.medium.com/building-a-fully-automated-medium-stats-pipeline-to-track-my-writing-performance-e76f791cdd59
P 19
marvin 13
🚀 18
s
I've enjoyed the read. Thank you for sharing it. Aside, the link URL for the second citation ("The startup data stack starter pack (2020)") seems to be a duplicate of the first cite.
f
@Steve Taylor thanks for the heads up about the link and sorry for the late reply. I adjusted the link and in case you still curious about the content here it is 🙂 https://dataform.co/blog/the-startup-data-stack-starter-pack Thanks!
👍 1
t
Hi @Felix Vemmer! Is your code available somewhere on Gihub, I’d really like to have a look into it! Thank you!
f
@Thomas Reynaud for now it’s in a private repo with some other automations. But happy to put something online. Is there any particualr part you are interested in? 🙂
t
Yes, we’re having trouble with structuring our Prefect projects, particularly how to avoid putting all the utility functions in the same .py file than the prefect flow. It always ends up being a huge file with everything in it and I don’t really like that so I wanted to see how you structured yours 🙂
Basically we have issues with relative imports 😄
f
Aaah I see, so for the project above I have unfortunately done the same… but for another project I approached it the following way. 1. Build a custom class that takes care of all the core functionality
2. Build a custom prefect task (subclass from the library)
3. Take this custom prefect subclass talk into the final script
Example for 1:
Copy code
from notion.client import NotionClient
import prefect
import logging
import pandas as pd


class Notion():

    def __init__(self, notion_token_v2, loger_type='prefect') -> None:

        if loger_type == 'prefect':
            self.logger = prefect.context.get("logger")
        else:
            self.logger = logging

        self.notion_token_v2 = notion_token_v2
        self.notion_client = self.create_notion_client()

    def create_notion_client(self):

        client = NotionClient(token_v2=self.notion_token_v2)

        <http://self.logger.info|self.logger.info>(
            f"Logged in as: {client.current_user.full_name}")

        return client

    def get_social_media_posts(self, notion_url):

        cv = self.notion_client.get_collection_view(notion_url)

        filter_params = {"filters": [
            {"filter": {"value": {"type": "relative", "value": "today"},
                        "operator": "date_is"}, "property": "jUaf"},
            {"filter": {"value": {"type": "exact", "value": False},
                        "operator": "checkbox_is"}, "property": "U<;g"}
        ], "operator": "and"}

        result = cv.build_query(filter=filter_params).execute()

        if len(result) > 0:
            result = result[0]

            <http://self.logger.info|self.logger.info>(
                f"Post for today: \nTitle: {result.title}\nLinkedIn Post Text: {result.linkedin_post_text}\nTwitter Post Text: {result.twitter_post_text}\nSuggested by: {result.content_found_by[0].full_name}")

        else:

            <http://self.logger.info|self.logger.info>("Nothing to post for today...")
            result = None

        return result
Example for 2.
Copy code
rom typing import Any
import pandas as pd
from core_automation.notion_client import Notion
from prefect.utilities.tasks import defaults_from_attrs


from prefect import Task


class ImportNotionLinkedInPosts(Task):

    def __init__(
        self,
        notion_token_v2: str = None,
        loger_type: str = None,
        df: pd.DataFrame = None,
        notion_url: str = None,
        **kwargs: Any
    ):

        self.notion_token_v2 = notion_token_v2
        self.loger_type = loger_type
        self.df = df
        self.notion_url = notion_url

        super().__init__(**kwargs)

    @defaults_from_attrs("notion_token_v2", "loger_type", "df", "notion_url")
    def run(
        self,
        notion_token_v2: str = None,
        loger_type: str = None,
        df: pd.DataFrame = None,
        notion_url: str = None,
    ):

        notion = Notion(
            notion_token_v2=notion_token_v2,
            loger_type=loger_type
        )

        notion.export_linkedin_posts(
            df=df,
            notion_url=notion_url
        )
3.
Copy code
import_notion_linkedin_posts = ImportNotionLinkedInPosts()

with Flow("LinkedIn Flow", run_config=LocalRun(),
          schedule=weekly_schedule
          ) as flow:

......
.....
.....

    import_notion_linkedin_posts(
        notion_token_v2=notion_token_v2,
        loger_type='prefect',
        df=liked_post_feed,
        notion_url=notion_url
    )
3 Views
Open in Slack
PreviousNext
Powered by