Feature/export

app/internal/agenda_events.py

@@ -1,3 +1,4 @@
+import datetime
 from datetime import date, timedelta
 from typing import Iterator, List, Optional, Union
 
@@ -59,11 +60,32 @@ def get_time_delta_string(start: date, end: date) -> str:
 
 
 def filter_dates(
-        events: List[Event], start: Optional[date],
-        end: Optional[date]) -> Iterator[Event]:
-    """filter events by a time frame."""
+        events: List[Event],
+        start: Union[None, date] = None,
+        end: Union[None, date] = None,
+) -> Iterator[Event]:
+    """Returns all events in a time frame.
 
-    yield from (
-        event for event in events
-        if start <= event.start.date() <= end
-    )
+    if "start_date" or "end_date" are None,
+    it will ignore that parameter when filtering.
+
+    for example:
+    if start_date = None and end_date = datetime.now().date,
+    then the function will return all events that ends before end_date.
+    """
+    start = start or datetime.date.min
+    end = end or datetime.date.max
+
+    for event in events:
+        if start <= event.start.date() <= end:
+            yield event
+
+
+def get_events_in_time_frame(
+        start_date: Union[date, None],
+        end_date: Union[date, None],
+        user_id: int, db: Session
+) -> Iterator[Event]:
+    """Yields all user's events in a time frame."""
+    events = get_all_user_events(db, user_id)
+    yield from filter_dates(events, start_date, end_date)

app/internal/export.py

@@ -0,0 +1,118 @@
+from datetime import datetime
+from typing import List
+
+import pytz
+from icalendar import Calendar, vCalAddress, vText
+from icalendar import Event as IcalEvent
+from sqlalchemy.orm import Session
+
+from app.config import DOMAIN, ICAL_VERSION, PRODUCT_ID
+from app.database.models import Event
+from app.internal.email import verify_email_pattern
+from app.routers.event import get_attendees_email
+
+
+def generate_id(event: Event) -> bytes:
+    """Creates an unique id."""
+    return (
+            str(event.id)
+            + event.start.strftime('%Y%m%d')
+            + event.end.strftime('%Y%m%d')
+            + f'@{DOMAIN}'
+    ).encode()
+
+
+def create_ical_calendar():
+    """Creates an ical calendar, and adds the required information"""
+    cal = Calendar()
+    cal.add('version', ICAL_VERSION)
+    cal.add('prodid', PRODUCT_ID)
+
+    return cal
+
+
+def add_optional(user_event, data):
+    """Adds an optional field if it exists."""
+    if user_event.location:
+        data.append(('location', user_event.location))
+
+    if user_event.content:
+        data.append(('description', user_event.content))
+
+    return data
+
+
+def create_ical_event(user_event):
+    """Creates an ical event, and adds the event information"""
+    ievent = IcalEvent()
+    data = [
+        ('organizer', add_attendee(user_event.owner.email, organizer=True)),
+        ('uid', generate_id(user_event)),
+        ('dtstart', user_event.start),
+        ('dtstamp', datetime.now(tz=pytz.utc)),
+        ('dtend', user_event.end),
+        ('summary', user_event.title),
+    ]
+
+    data = add_optional(user_event, data)
+
+    for param in data:
+        ievent.add(*param)
+
+    return ievent
+
+
+def add_attendee(email, organizer=False):
+    """Adds an attendee to the event."""
+    attendee = vCalAddress(f'MAILTO:{email}')
+    if organizer:
+        attendee.params['partstat'] = vText('ACCEPTED')
+        attendee.params['role'] = vText('CHAIR')
+    else:
+        attendee.params['partstat'] = vText('NEEDS-ACTION')
+        attendee.params['role'] = vText('PARTICIPANT')
+
+    return attendee
+
+
+def add_attendees(
+        ievent: IcalEvent,
+        attendees: List[str],
+):
+    """Adds attendees for the event."""
+    for email in attendees:
+        if verify_email_pattern(email):
+            ievent.add(
+                'attendee',
+                add_attendee(email),
+                encode=0
+            )
+
+    return ievent
+
+
+def event_to_ical(user_event: Event, attendees: List[str]) -> bytes:
+    """Returns an ical event, given an
+    "Event" instance and a list of email."""
+    ical = create_ical_calendar()
+    ievent = create_ical_event(user_event)
+    ievent = add_attendees(ievent, attendees)
+    ical.add_component(ievent)
+
+    return ical.to_ical()
+
+
+def export_calendar(session: Session, events: List[Event]) -> bytes:
+    """Returns an icalendar, given an list of
+    "Event" instances and a list of email."""
+    ical = create_ical_calendar()
+    for event in events:
+        ievent = create_ical_event(event)
+
+        attendees = get_attendees_email(session, event)
+        attendees.remove((event.owner.email,))
+
+        ievent = add_attendees(ievent, attendees)
+        ical.add_component(ievent)
+
+    return ical.to_ical()

app/main.py

@@ -6,6 +6,7 @@
 from app.database import engine, models
 from app.dependencies import get_db, logger, MEDIA_PATH, STATIC_PATH, templates
 from app.internal import daily_quotes, json_data_loader
+
 from app.internal.languages import set_ui_language
 from app.routers.salary import routes as salary
 
@@ -35,8 +36,9 @@ def create_tables(engine, psql_environment):
 
 
 from app.routers import (  # noqa: E402
+
     agenda, calendar, categories, celebrity, currency, dayview,
-    email, event, four_o_four, invitation, profile, search,
+    email, event, export, four_o_four, invitation, profile, search,
     weekview, telegram, whatsapp,
 )
 
@@ -52,6 +54,7 @@ def create_tables(engine, psql_environment):
     weekview.router,
     email.router,
     event.router,
+    export.router,
     four_o_four.router,
     invitation.router,
     profile.router,

app/routers/calendar_grid.py

@@ -8,7 +8,7 @@
 
 MONTH_BLOCK: int = 6
 
-locale.setlocale(locale.LC_TIME, ("en", "UTF-8"))
+locale.setlocale(locale.LC_ALL, "en_US.UTF-8")
 
 
 class Day:

app/routers/event.py

@@ -14,6 +14,7 @@
 from app.internal.event import (
     get_invited_emails, get_messages, get_uninvited_regular_emails,
     raise_if_zoom_link_invalid,
+
 )
 from app.internal.emotion import get_emotion
 from app.internal.utils import create_model
@@ -243,6 +244,13 @@ def sort_by_date(events: List[Event]) -> List[Event]:
     return sorted(temp, key=attrgetter('start'))
 
 
+def get_attendees_email(session: Session, event: Event):
+    return (
+        session.query(User.email).join(UserEvent)
+        .filter(UserEvent.events == event).all()
+    )
+
+
 def get_participants_emails_by_event(db: Session, event_id: int) -> List[str]:
     """Returns a list of all the email address of the event invited users,
         by event id."""

app/routers/export.py

@@ -1,103 +1,37 @@
-from datetime import datetime
-from typing import List
-
-from icalendar import Calendar, Event, vCalAddress, vText
-import pytz
-
-from app.config import DOMAIN, ICAL_VERSION, PRODUCT_ID
-from app.database.models import Event as UserEvent
-
-
-def generate_id(event: UserEvent) -> bytes:
-    """Creates an unique id."""
-
-    return (
-            str(event.id)
-            + event.start.strftime('%Y%m%d')
-            + event.end.strftime('%Y%m%d')
-            + f'@{DOMAIN}'
-    ).encode()
-
-
-def create_ical_calendar():
-    """Creates an ical calendar,
-    and adds the required information"""
-
-    cal = Calendar()
-    cal.add('version', ICAL_VERSION)
-    cal.add('prodid', PRODUCT_ID)
-
-    return cal
-
-
-def add_optional(user_event, data):
-    """Adds an optional field if it exists."""
-
-    if user_event.location:
-        data.append(('location', user_event.location))
-
-    if user_event.content:
-        data.append(('description', user_event.content))
-
-    return data
-
-
-def create_ical_event(user_event):
-    """Creates an ical event,
-    and adds the event information"""
-
-    ievent = Event()
-    data = [
-        ('organizer', add_attendee(user_event.owner.email, organizer=True)),
-        ('uid', generate_id(user_event)),
-        ('dtstart', user_event.start),
-        ('dtstamp', datetime.now(tz=pytz.utc)),
-        ('dtend', user_event.end),
-        ('summary', user_event.title),
-    ]
-
-    data = add_optional(user_event, data)
-
-    for param in data:
-        ievent.add(*param)
-
-    return ievent
-
-
-def add_attendee(email, organizer=False):
-    """Adds an attendee to the event."""
-
-    attendee = vCalAddress(f'MAILTO:{email}')
-    if organizer:
-        attendee.params['partstat'] = vText('ACCEPTED')
-        attendee.params['role'] = vText('CHAIR')
-    else:
-        attendee.params['partstat'] = vText('NEEDS-ACTION')
-        attendee.params['role'] = vText('PARTICIPANT')
-
-    return attendee
-
-
-def add_attendees(ievent, attendees: list):
-    """Adds attendees for the event."""
-
-    for email in attendees:
-        ievent.add(
-            'attendee',
-            add_attendee(email),
-            encode=0
-        )
-
-    return ievent
-
-
-def event_to_ical(user_event: UserEvent, attendees: List[str]) -> bytes:
-    """Returns an ical event, given an
-    "UserEvent" instance and a list of email."""
-
-    ical = create_ical_calendar()
-    ievent = create_ical_event(user_event)
-    ievent = add_attendees(ievent, attendees)
-    ical.add_component(ievent)
-
-    return ical.to_ical()
+from datetime import date
+from io import BytesIO
+from typing import Union
+
+from fastapi import APIRouter, Depends
+from sqlalchemy.orm import Session
+from starlette.responses import StreamingResponse
+
+from app.dependencies import get_db
+from app.internal.agenda_events import get_events_in_time_frame
+from app.internal.export import export_calendar
+
+router = APIRouter(
+    prefix="/export",
+    tags=["export"],
+    responses={404: {"description": "Not found"}},
+)
+
+
+@router.get("/")
+def export(
+        start_date: Union[date, str],  # date or an empty string
+        end_date: Union[date, str],
+        db: Session = Depends(get_db),
+) -> StreamingResponse:
+
+    user_id = 1
+    events = get_events_in_time_frame(start_date, end_date, user_id, db)
+    file = BytesIO(export_calendar(db, list(events)))
+    return StreamingResponse(
+        content=file,
+        media_type="text/calendar",
+        headers={
+            # change filename to "pylendar.ics"
+            "Content-Disposition": "attachment;filename=pylendar.ics"
+        },
+    )

app/routers/share.py

@@ -4,7 +4,7 @@
 
 from app.database.models import Event, Invitation, UserEvent
 from app.internal.utils import save
-from app.routers.export import event_to_ical
+from app.internal.export import event_to_ical
 from app.routers.user import does_user_exist, get_users
 
 

app/templates/profile.html

@@ -217,7 +217,13 @@ <h6 class="card-title text-center mb-1">{{ user.full_name }}</h6>
                 </li>
               {% endif %}
               <li class="list-group-item list-group-item-action no-border">
-                <a class="text-decoration-none text-secondary" href="#">Export my calendar</a>
+                <form method="GET" action="{{ url_for('export') }}">
+                    <label for="start_date">{{ gettext("From") }}</label><br>
+                    <input class="filter" type="date" id="start_date" name="start_date"><br>
+                    <label for="end_date">{{ gettext("To") }}</label><br>
+                    <input class="filter" type="date" id="end_date" name="end_date"><br>
+                    <input class="filter" type="submit" value="{{ gettext('Export') }}">
+                </form>
               </li>
               <li class="list-group-item list-group-item-action no-border">
                 <a class="text-decoration-none text-secondary" href="{{ url_for('import_holidays') }}">Add holidays to calendar</a>

tests/event_fixture.py

@@ -89,7 +89,7 @@ def old_event(sender: User, session: Session) -> Event:
         db=session,
         title='event 6',
         start=today_date - timedelta(days=5),
-        end=today_date,
+        end=today_date - timedelta(days=1),
         content='test event',
         owner_id=sender.id,
     )